spark 通过 RDD 从HDFS文件加载JSON文件到sql表

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了spark 通过 RDD 从HDFS文件加载JSON文件到sql表，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含4398字，纯文字阅读大概需要7分钟。

内容图文

RDD定义

RDD全称是Resilient Distributed Dataset, 是spark的核心抽象层，通过它可以读取多种文件，这里演示如何读取hdfs文件。所有spark的工作都是发生在RDD上，比如创建新的RDD，转换已有的RDD，对现有的RDD计算求得结果。

RDD在spark中是不可变的(immutable)对象集合，RDD可以被划分成多个分区，存放在不同的节点。

创建RDD

有两种方法，一种是加载外部的数据集，比如下面加载HDFS的文件, 运行在scalar-shell中：

val textFile = sc.textFile("hdfs://namenode-host:9000/input/dean/obd_hdfs-writer-4-9-1447126914492.log")
textFile.count()
res1: Long = 3574 

另一种方法是在driver program中用SparkContext的paralleize方法。这里暂时不讨论。

读取JSON文件

上面的log文件内容其实是json格式的，所以可以换种读法：

scala> val sqlContext = new org.apache.spark.sql.SQLContext(sc)
sqlContext: org.apache.spark.sql.SQLContext = org.apache.spark.sql.SQLContext@2f92b5a1

scala> val path = "hdfs://namenode-host:9000/input/dean/obd_hdfs-writer-4-9-1447126914492.log"
path: String = hdfs://namenode-host:9000/input/dean/obd_hdfs-writer-4-9-1447126914492.log

scala> val c = sqlContext.read.json(path)
c: org.apache.spark.sql.DataFrame = [data: struct<client_version:bigint,corp_id:string,east:bigint,ext_o_latitude:double,ext_o_longitude:double,gps_num:array<struct<east:bigint,gps_num:bigint,gpstime:bigint,latitude:double,longitude:double,msg_id:bigint,msg_length:bigint,msg_type:bigint,north:bigint,terminal:string,tsp_obd_n900_head:array<bigint>>>,gpstime:bigint,heading:bigint,k:string,latitude:double,longitude:double,msg_id:bigint,msg_length:bigint,msg_type:bigint,north:bigint,syn_type:bigint,systime_driverStorage:bigint,systime_listenerserver:bigint,target_id:string,target_name:string,terminal:string,terminal_id:string,terminal_status_desc:string,tsp_obd_n900_head:array<bigint>,type:bigint,update_time:bigint>, driverName: string, type: string]

scala> c.printSchema()
root
 |-- data: struct (nullable = true)
 |    |-- client_version: long (nullable = true)
 |    |-- corp_id: string (nullable = true)
 |    |-- east: long (nullable = true)
 |    |-- ext_o_latitude: double (nullable = true)
 |    |-- ext_o_longitude: double (nullable = true)
 |    |-- gps_num: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- east: long (nullable = true)
 |    |    |    |-- gps_num: long (nullable = true)
 |    |    |    |-- gpstime: long (nullable = true)
 |    |    |    |-- latitude: double (nullable = true)
 |    |    |    |-- longitude: double (nullable = true)
 |    |    |    |-- msg_id: long (nullable = true)
 |    |    |    |-- msg_length: long (nullable = true)
 |    |    |    |-- msg_type: long (nullable = true)
 |    |    |    |-- north: long (nullable = true)
 |    |    |    |-- terminal: string (nullable = true)
 |    |    |    |-- tsp_obd_n900_head: array (nullable = true)
 |    |    |    |    |-- element: long (containsNull = true)
 |    |-- gpstime: long (nullable = true)
 |    |-- heading: long (nullable = true)
 |    |-- k: string (nullable = true)
 |    |-- latitude: double (nullable = true)
 |    |-- longitude: double (nullable = true)
 |    |-- msg_id: long (nullable = true)
 |    |-- msg_length: long (nullable = true)
 |    |-- msg_type: long (nullable = true)
 |    |-- north: long (nullable = true)
 |    |-- syn_type: long (nullable = true)
 |    |-- systime_driverStorage: long (nullable = true)
 |    |-- systime_listenerserver: long (nullable = true)
 |    |-- target_id: string (nullable = true)
 |    |-- target_name: string (nullable = true)
 |    |-- terminal: string (nullable = true)
 |    |-- terminal_id: string (nullable = true)
 |    |-- terminal_status_desc: string (nullable = true)
 |    |-- tsp_obd_n900_head: array (nullable = true)
 |    |    |-- element: long (containsNull = true)
 |    |-- type: long (nullable = true)
 |    |-- update_time: long (nullable = true)
 |-- driverName: string (nullable = true)
 |-- type: string (nullable = true)

转换成表格

现在来写入到临时表obd中，并遍历该表的内容：

c.registerTempTable("obd")
val set = sqlContext.sql("select * from obd")
set.collect().foreach(println)

会自动将JSON的树形结构拉平，不管好不好，至少是个能用的表。

这是一个程序和sql混在一起使用的模式，有点意思，不过还有些缺憾。既然是程序，就需要自动补全等功能，spark-shell没有提供。

版权声明：本文为博主原创文章，未经博主允许不得转载。

spark 通过 RDD 从HDFS文件加载JSON文件到sql表

标签：spark   hdfs   

本文系统来源：http://blog.csdn.net/csfreebird/article/details/49766635

内容总结

以上是互联网集市为您收集整理的spark 通过 RDD 从HDFS文件加载JSON文件到sql表全部内容，希望文章能够帮你解决spark 通过 RDD 从HDFS文件加载JSON文件到sql表所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

---