火花流 xml 文件答案

【问题标题】：Spark Streaming xml files火花流 xml 文件
【发布时间】：2017-04-02 11:11:15
【问题描述】：

我需要处理流入 S3 文件夹的 xml 文件。目前，我已将其实现如下。

首先，使用 Spark 的 fileStream 读取文件

val data = ssc.fileStream[LongWritable, Text, TextInputFormat]("s3://myfolder/",(t: org.apache.hadoop.fs.Path) => true, newFilesOnly = true, hadoopConf).map(_._2.toString())

对于每个RDD，检查是否有文件被读取

if (data.count() !=0)

将字符串写入新的 HDFS 目录

data.coalesce(1).saveAsTextFile(sdir);

创建一个从上面的HDFS目录读取的Dataframe

val loaddata = sqlContext.read.format("com.databricks.spark.xml").option("rowTag", "Trans").load(sdir)

对Dataframe进行一些处理并保存为JSON

loaddata.write.mode("append").json("s3://mybucket/somefolder")

不知何故，我觉得上述方法非常低效，坦率地说相当学校孩子气。有更好的解决方案吗？任何帮助将不胜感激。

一个后续问题：如何操作数据框中的字段（不是列）？我有一个非常复杂的嵌套 xml，当我使用上述方法时，我得到一个包含 9 列和 50 个奇数内部结构数组的数据框。这很好，除了需要修剪某些字段名称。有没有办法在不爆炸数据框的情况下实现这一点，因为我需要再次构建相同的结构？

【问题讨论】：

标签： xml apache-spark spark-structured-streaming

【解决方案1】：

如果您使用 Spark 2.0，您也许可以使其与结构化流式传输一起使用：

val inputDF = spark.readStream.format("com.databricks.spark.xml")
  .option("rowTag", "Trans")
  .load(path)

【讨论】：

非常感谢。我的目标环境是带有 Spark 2.0.1 的 EMR 堆栈。我会在 EMR 盒子上尝试你的建议。
如果您同意上述解决方案，请投票/接受。
val inputDF = spark.readStream.format("com.databricks.spark.xml") .option("rowTag", "Trans") .load(path) 以上解决方案似乎不适用于 spark 2.X