• linkedu视频
  • 平面设计
  • 电脑入门
  • 操作系统
  • 办公应用
  • 电脑硬件
  • 动画设计
  • 3D设计
  • 网页设计
  • CAD设计
  • 影音处理
  • 数据库
  • 程序设计
  • 认证考试
  • 信息管理
  • 信息安全
菜单
linkedu.com
导航菜单
  • 网页制作
  • 数据库
  • 程序设计
  • 操作系统
  • CMS教程
  • 游戏攻略
  • 脚本语言
  • 平面设计
  • 软件教程
  • 网络安全
  • 电脑知识
  • 服务器
  • 视频教程
  • windows
  • 服务器硬件
  • 服务器运维
  • 云计算
  • 虚拟化
  • IIS教程
  • Linux
  • Apache
  • Ftp
  • DNS
  • Nginx
您的位置:首页 > 服务器 >云计算 > Spark SQL and DataFrame Guide(1.4.1)——之DataFrames,sparkdataframe

Spark SQL and DataFrame Guide(1.4.1)——之DataFrames,sparkdataframe

作者:网友 字体:[增加 减小] 来源:互联网

本文主要包含Spark SQL and DataFrame Guide(1.4.1)——之DataFrames,sparkdataframe等服务器相关知识,网友希望可以进行参考

Spark SQL and DataFrame Guide(1.4.1)——之DataFrames,sparkdataframe


Spark SQL是处理结构化数据的Spark模块。它提供了DataFrames这种编程抽象,同时也可以作为分布式SQL查询引擎使用。

DataFrames

DataFrame是一个带有列名的分布式数据集合。等同于一张关系型数据库中的表或者R/Python中的data frame,不过在底层做了很多优化;我们可以使用结构化数据文件、Hive tables,外部数据库或者RDDS来构造DataFrames。

1. 开始入口:

入口需要从SQLContext类或者它的子类开始,当然需要使用SparkContext创建SQLContext;这里我们使用pyspark(已经自带了SQLContext即sc):

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

还可以使用HiveContext,它可以提供比SQLContext更多的功能,例如可以使用更完整的HiveQL解析器写查询,使用Hive UDFs,从Hive表中读取数据等。使用HiveContext并不需要安装hive,Spark默认将HiveContext单独打包避免对hive过多的依赖

2.创建DataFrames
使用JSON文件创建:

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

df = sqlContext.read.json("examples/src/main/resources/people.json")

# Displays the content of the DataFrame to stdout
df.show()

注意:
这里你可能需要将文件存入HDFS(这里的文件在Spark安装目录中,1.4版本)

hadoop fs -mkdir examples/src/main/resources/
hadoop fs -put /appcom/spark/examples/src/main/resources/*         /user/hdpuser/examples/src/main/resources/

3.DataFrame操作

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

# Create the DataFrame
df = sqlContext.read.json("examples/src/main/resources/people.json")

# Show the content of the DataFrame
df.show()
## age  name
## null Michael
## 30   Andy
## 19   Justin

# Print the schema in a tree format
df.printSchema()
## root
## |-- age: long (nullable = true)
## |-- name: string (nullable = true)

# Select only the "name" column
df.select("name").show()
## name
## Michael
## Andy
## Justin

# Select everybody, but increment the age by 1
df.select(df['name'], df['age'] + 1).show()
## name    (age + 1)
## Michael null
## Andy    31
## Justin  20

# Select people older than 21
df.filter(df['age'] > 21).show()
## age name
## 30  Andy

# Count people by age
df.groupBy("age").count().show()
## age  count
## null 1
## 19   1
## 30   1

4.使用编程运行SQL查询
SQLContext可以使用编程运行SQL查询并返回DataFrame。

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.sql("SELECT * FROM table")

5.和RDD交互

将RDD转换成DataFrames有两种方法:

  • 利用反射来推断包含特定类型对象的RDD的schema。这种方法会简化代码并且在你已经知道schema的时候非常适用。
  • 使用编程接口,构造一个schema并将其应用在已知的RDD上。

一、利用反射推断Schema
Spark SQL能够将含Row对象的RDD转换成DataFrame,并推断数据类型。通过将一个键值对(key/value)列表作为kwargs传给Row类来构造Rows。key定义了表的列名,类型通过看第一列数据来推断。(所以这里RDD的第一列数据不能有缺失)未来版本中将会通过看更多数据来推断数据类型,像现在对JSON文件的处理一样。

# sc is an existing SparkContext.
from pyspark.sql import SQLContext, Row
sqlContext = SQLContext(sc)

# Load a text file and convert each line to a Row.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: Row(name=p[0], age=int(p[1])))

# Infer the schema, and register the DataFrame as a table.
schemaPeople = sqlContext.createDataFrame(people)
schemaPeople.registerTempTable("people")

# SQL can be run over DataFrames that have been registered as a table.
teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19")

# The results of SQL queries are RDDs and support all the normal RDD operations.
teenNames = teenagers.map(lambda p: "Name: " + p.name)
for teenName in teenNames.collect():
  print teenName

二、编程指定Schema
通过编程指定Schema需要3步:

# Import SQLContext and data types
from pyspark.sql import SQLContext
from pyspark.sql.types import *

# sc is an existing SparkContext.
sqlContext = SQLContext(sc)

# Load a text file and convert each line to a tuple.
lines = sc.textFile("examples/src/main/resources/people.txt")
parts = lines.map(lambda l: l.split(","))
people = parts.map(lambda p: (p[0], p[1].strip()))

# The schema is encoded in a string.
schemaString = "name age"

fields = [StructField(field_name, StringType(), True) for field_name in sch
  


 
分享到:QQ空间新浪微博腾讯微博微信百度贴吧QQ好友复制网址打印

您可能想查找下面的文章:

  • Spark SQL and DataFrame Guide(1.4.1)——之DataFrames,sparkdataframe

相关文章

  • 给大数据文件的每一行产生唯一的id,数据一行id
  • Error: Failed to launch instance &quot;win7&quot;: Please try again later [Error: No valid host was found. ].,win7valid
  • 用Hello World校验Docker的安装,hellodocker
  • OSTC 2015,ostc2015
  • Hbase namespace问题,hbasenamespace问题
  • 使用 ElasticSearch + LogStash + Kibana 来可视化网络流量,kibanaelasticsearch
  • spark core源码分析16 Shuffle详解-读流程,sparkshuffle
  • HIVE以及OOZIE添加第三方JAR包的方法,hiveoozie
  • Hadoop 使用常见问题,hadoop使用常见问题
  • [Hive]HBaseIntegration:通过Hive读写HBase,

文章分类

  • windows
  • 服务器硬件
  • 服务器运维
  • 云计算
  • 虚拟化
  • IIS教程
  • Linux
  • Apache
  • Ftp
  • DNS
  • Nginx

最近更新的内容

    • Openstack中为虚拟机使用CDROM光驱设备,openstackcdrom
    • Hadoop文件解压缩,
    • Spark MLlib LDA主题模型(1),mlliblda
    • pig里面没有if:不能判断一个条件后决定一个执行步骤,pigif
    • AMQP server on c ontroller:5672 is unreachable: [Errno 113] EHOSTUNREACH. Trying again in 2 seconds.,ehostunreach
    • HIVE动态分区实战,hive动态实战
    • spark core源码分析8 从简单例子看transformation,transformation
    • 【Spark】弹性分布式数据集RDD概述,sparkrdd概述
    • MapReduce编程之实现多表关联,mapreduce编程关联
    • Hive 和普通关系数据库的异同,hive关系数据库

关于我们 - 联系我们 - 免责声明 - 网站地图

©2020-2025 All Rights Reserved. linkedu.com 版权所有