内容简介：Spark Sql的核心对象DataFrame提供了Sql查询的能力，极大的方便了数据的查询和统计。如果要对DataFrame进行Sql查询，需要指定Scheme信息。而文本文件往往缺少Scheme信息，这篇文章将演示3种从文本文件创建DataFrame，并设置其Scheme的方式。本文用来做示范的文件，是一个简化的手机端的事件日志，以“|”分隔，其数据格式为：一般我们在代码里习惯用英文，翻译一下就是：
Spark Sql的核心对象DataFrame提供了 Sql 查询的能力，极大的方便了数据的查询和统计。如果要对DataFrame进行Sql查询，需要指定Scheme信息。而文本文件往往缺少Scheme信息，这篇文章将演示3种从文本文件创建DataFrame，并设置其Scheme的方式。
范例文件介绍
本文用来做示范的文件，是一个简化的手机端的事件日志，以“|”分隔，其数据格式为：
客户端ip|事件时间|事件标识|机器码|操作系统|设备型号
一般我们在代码里习惯用英文，翻译一下就是：
ip|date|event|machine|os|device
事件日志的名称为 events.log，内容如下：
49.77.134.137|1536544453|touxiang_1267|868385038260432|android 7.1.1 R11s_11_A.25_180706|OPPO R11s 223.104.145.59|1536544452|icon_frame_314|865902037967669|android 7.1.1 NMF26F|MI MAX 2 222.93.202.10|1536544344|btn_speak_30|867701035125231|android 7.1.1 NMF26X release-keys|vivo X20Plus A 222.93.202.10|1536544346|img_sound_293|867701035125231|android 7.1.1 NMF26X release-keys|vivo X20Plus A 222.93.202.10|1536544437|btn_hui_fu_1997|867701035125231|android 7.1.1 NMF26X release-keys|vivo X20Plus A 117.91.166.96|1536544419|btn_speak_30|864088038729653|android 6.0.1 R9sPlus_11_A.11_180515|OPPO R9s Plus 117.91.166.96|1536544426|btn_speak_30|864088038729653|android 6.0.1 R9sPlus_11_A.11_180515|OPPO R9s Plus 117.91.166.96|1536544433|btn_speak_30|864088038729653|android 6.0.1 R9sPlus_11_A.11_180515|OPPO R9s Plus
先将这个文件从当前目录（~/downloads/events.log）上传到hdfs的/test文件夹下，方便后续测试：
# hdfs dfs -put ~/downloads/events.log /test
方式1：从RDD创建
第一种方式先创建RDD，然后将RDD的行转换为Row对象，再根据Row去推断Schema。也可以转换为元组，再明确指定Schema。
# coding=utf-8
from pyspark.sql import SparkSession
from pyspark import Row
from pyspark.sql import types as T
spark = SparkSession.builder\
.master("local[*]")\
.appName("test.dataframe")\
.getOrCreate()
sc = spark.sparkContext
linesRDD = sc.textFile("/test/events.log")
partsRDD = linesRDD.map(lambda x: x.split("|"))
# 方法1：创建ROW对象，根据ROW来创建Schema，自动推断字段类型
rowsRDD = partsRDD.map(lambda x:Row(ip=x[0],date=int(x[1]),event=x[2], machine=x[3], os=x[4], device=x[5]))
df = spark.createDataFrame(rowsRDD)
# 方法2：通过Schema字符串创建
# rowsRDD = partsRDD.map(lambda x:(x[0], int(x[1]), x[2], x[3], x[4], x[5]))
# schema = "ip: string, date: long, event: string, machine: string, os: string, device: string"
# df = spark.createDataFrame(rowsRDD, schema)
# 方法3：通过StructType对象创建
# rowsRDD = partsRDD.map(lambda x:(x[0], int(x[1]), x[2], x[3], x[4], x[5]))
# schema = T.StructType([
# T.StructField("ip", T.StringType(), True),
# T.StructField("date", T.LongType(), False),
# T.StructField("event", T.StringType()),
# T.StructField("machine", T.StringType(), True),
# T.StructField("os", T.StringType(), True),
# T.StructField("device", T.StringType(), True),
# ])
# df = spark.createDataFrame(rowsRDD, schema)
df.printSchema()
df.createOrReplaceTempView("events")
df = spark.sql("select * from events where device like '%OPPO%'")
df.show()
print "success"
这里采用了三种给RDD赋予Scheme的方法。最复杂(繁琐)的是第3种，但是第3种可以指定Field是否可以为Null。
结果如下（3种方法的结果大同小异）：
# $SPARK_HOME/bin/spark-submit /data/pyjobs/sample/dataframe-rdd.py root |-- ip: string (nullable = true) |-- date: long (nullable = true) |-- event: string (nullable = true) |-- machine: string (nullable = true) |-- os: string (nullable = true) |-- device: string (nullable = true) +-------------+----------+-------------+---------------+--------------------+-------------+ | ip| date| event| machine| os| device| +-------------+----------+-------------+---------------+--------------------+-------------+ |49.77.134.137|1536544453|touxiang_1267|868385038260432|android 7.1.1 R11...| OPPO R11s| |117.91.166.96|1536544419| btn_speak_30|864088038729653|android 6.0.1 R9s...|OPPO R9s Plus| |117.91.166.96|1536544426| btn_speak_30|864088038729653|android 6.0.1 R9s...|OPPO R9s Plus| |117.91.166.96|1536544433| btn_speak_30|864088038729653|android 6.0.1 R9s...|OPPO R9s Plus| +-------------+----------+-------------+---------------+--------------------+-------------+ success
方式2：从DataFrame转换
从DataFrame转换，就是SparkSession的read方法，直接将文件读取为DataFrame，此时的Schema就一个string类型的字段，默认名为value，接着再对此DataFrame进行转换。
需要注意：pyspark.sql.functions.split函数，其接收的是一个正则表达式，而“|”在正则表达式中是一个特殊字符，因此需要用“\”进行转义。
# coding=utf-8
from pyspark.sql import SparkSession
from pyspark.sql import types as T
from pyspark.sql import functions as F
spark = SparkSession.builder\
.master("local[*]")\
.appName("test.dataframe")\
.getOrCreate()
# step1. 原始数据
df = spark.read.text("/test/events.log")
df.printSchema()
df.show()
# step2. 转成了数组
df = df.select(F.split(df["value"], '\|').alias("cols"))
df.printSchema()
df.show()
# step3. 将数组字段转成了列
cols = df["cols"]
df = df.withColumn("ip", cols.getItem(0)) \
.withColumn("date", cols.getItem(1).cast(T.LongType())) \
.withColumn("event", cols.getItem(2)) \
.withColumn("machine", cols.getItem(3)) \
.withColumn("os", cols.getItem(4)) \
.withColumn("device", cols.getItem(5)) \
.drop("cols")
df.printSchema()
df.createOrReplaceTempView("events")
df = spark.sql("select * from events where device like '%OPPO%'")
df.show()
print "success"
输出的结果如下：
# $SPARK_HOME/bin/spark-submit /data/pyjobs/sample/dataframe-df.py root |-- value: string (nullable = true) +--------------------+ | value| +--------------------+ |49.77.134.137|153...| |223.104.145.59|15...| |222.93.202.10|153...| |222.93.202.10|153...| |222.93.202.10|153...| |117.91.166.96|153...| |117.91.166.96|153...| |117.91.166.96|153...| +--------------------+ root |-- cols: array (nullable = true) | |-- element: string (containsNull = true) +--------------------+ | cols| +--------------------+ |[49.77.134.137, 1...| |[223.104.145.59, ...| |[222.93.202.10, 1...| |[222.93.202.10, 1...| |[222.93.202.10, 1...| |[117.91.166.96, 1...| |[117.91.166.96, 1...| |[117.91.166.96, 1...| +--------------------+ root |-- ip: string (nullable = true) |-- date: long (nullable = true) |-- event: string (nullable = true) |-- machine: string (nullable = true) |-- os: string (nullable = true) |-- device: string (nullable = true) +-------------+----------+-------------+---------------+--------------------+-------------+ | ip| date| event| machine| os| device| +-------------+----------+-------------+---------------+--------------------+-------------+ |49.77.134.137|1536544453|touxiang_1267|868385038260432|android 7.1.1 R11...| OPPO R11s| |117.91.166.96|1536544419| btn_speak_30|864088038729653|android 6.0.1 R9s...|OPPO R9s Plus| |117.91.166.96|1536544426| btn_speak_30|864088038729653|android 6.0.1 R9s...|OPPO R9s Plus| |117.91.166.96|1536544433| btn_speak_30|864088038729653|android 6.0.1 R9s...|OPPO R9s Plus| +-------------+----------+-------------+---------------+--------------------+-------------+ success
方式3：使用csv方法
最后一种就是使用csv方法了，可以直接指定分隔符和schema：
需要注意：这里的schema字符串格式，和SparkSession.CreateDataFrame方法中接收的schema字符串格式略有差异，少了“:”。
# coding=utf-8
from pyspark.sql import SparkSession
spark = SparkSession.builder\
.master("local[*]")\
.appName("test.dataframe")\
.getOrCreate()
schema = "ip string, date long, event string, machine string, os string, device string"
df = spark.read.csv("/test/events.log", sep="|", schema=schema)
df.printSchema()
df.createOrReplaceTempView("events")
df = spark.sql("select * from events where device like '%OPPO%'")
df.show()
print "success"
其运行结果和方式1是类似的，就不再演示了。
总结
不是后缀名是csv的文件就是csv文件，csv文件本意是以逗号分隔的文件，扩展一下，就是以某字符进行分隔的文件都可以视为csv文件。因此，对于上面的事件日志文件，也可以视为csv文件，那么显然，使用上面的方式3是最方便快捷的了。我本人最开始用的是方式1，兜兜转转才发现了方式3，文档还是看得不够仔细呀。总的来说，Spark的文档在几个开源大数据项目中，算是最优秀的一个了。有时间就看看说不定就会有新的收获。
感谢阅读，希望这篇文章能给你带来帮助！
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持 码农网
猜你喜欢:
- 游戏制作之路（28）创建界面文本显示
- 在Swift 5中从原始文本创建字符串
- 【PS-3D功能探索】如何在ps内创建膨胀的3D文本效果
- 基于标签特定文本表示的文本多标签分类
- 论文浅尝 | 通过文本到文本神经问题生成的机器理解
- 文本挖掘从小白到精通（三）---主题模型和文本数据转换
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
PHP高级开发技术与应用
曹铁群、孙一江、张永学 / 清华大学出版社 / 2002-5-1 / 32.00
作为一本介绍PHP高级开发技术的书籍，本书并不像一般介绍PHP语言的书籍那样讲述大量的语法规则，罗列大量的函数，而是着眼于PHP在Web中的实际应用，特别是PHP对最新技术的支持，比如WAP技术、XML技术等。 本书涉及到的内容主要有：高级环境配置、高级语法和应用、正则表达式、面向对象技术、高级图像技术、用PHPLIB实现模板的处理、用PHPDoc实现文档的自动生成、PHP与组件技术、PH......一起来看看 《PHP高级开发技术与应用》 这本书的介绍吧!
