scala - 无法从Sqoop创建的Spark中的序列文件创建数据框

Question

我想读取 orders 数据并从中创建 RDD，它作为 sequence 文件存储在 cloudera 的 hadoop fs 中 虚拟机。以下是我的步骤:

1) 将订单数据导入为序列文件:

sqoop import --connect jdbc:mysql://localhost/retail_db --username retail_dba --password cloudera  --table orders -m 1 --target-dir /ordersDataSet --as-sequencefile   

2)在spark scala中读取文件:

星火 1.6

val sequenceData=sc.sequenceFile("/ordersDataSet",classOf[org.apache.hadoop.io.Text],classOf[org.apache.hadoop.io.Text]).map(rec => rec.toString())  

3) 当我尝试从上面的 RDD 读取数据时，它抛出以下错误:

Caused by: java.io.IOException: WritableName can't load class: orders
    at org.apache.hadoop.io.WritableName.getClass(WritableName.java:77)
    at org.apache.hadoop.io.SequenceFile$Reader.getValueClass(SequenceFile.java:2108)
    ... 17 more
Caused by: java.lang.ClassNotFoundException: Class orders not found
    at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2185)
    at org.apache.hadoop.io.WritableName.getClass(WritableName.java:75)
    ... 18 more

不知道为什么说找不到订单。我哪里错了？

我也引用了这两个链接的代码，但运气不好:
1) Refer sequence part
2) Refer step no. 8

Answer 1

sqoop 与它关系不大，这里是一个更现实的场景示例，其中 saveAsSequenceFile 始终假定 k、v 对 - 这可能对您有帮助:

import org.apache.hadoop.io._

val RDD = sc.parallelize( List( (1, List("A", "B")) , (2, List("B", "C")) , (3, List("C", "D", "E")) ) )
val RDD2 = RDD.map(x => (x._1, x._2.mkString("/")))
RDD2.saveAsSequenceFile("/rushhour/seq-directory/2")

val sequence_data = sc.sequenceFile("/rushhour/seq-directory/*", classOf[IntWritable], classOf[Text])
                  .map{case (x, y) => (x.get(), y.toString().split("/")(0), y.toString().split("/")(1))}

sequence_data.collect

返回:

res20: Array[(Int, String, String)] = Array((1,A,B), (2,B,C), (3,C,D), (1,A,B), (2,B,C), (3,C,D))

我不确定您是想要 RDD 还是 DF，但是将 RDD 转换为 DF 当然是微不足道的。