从 ES 获取数据并作为 Avro (Spark) 保存到 HDFS答案

【问题标题】：Fetch data from ES and save to HDFS as Avro (Spark)从 ES 获取数据并作为 Avro (Spark) 保存到 HDFS
【发布时间】：2017-10-10 07:29:35
【问题描述】：

我是 Spark 和 Scala 的新手，所以我有一个非常基本的问题。我有 Spark 1.5.2 和 Scala 2.10.4。我创建的配置允许我在使用--config 命令启动spark-shell 时直接从ES 获取一些数据作为JSON；我还导入了elasticsearch-hadoop 的--jar。启动 Spark 后，我执行了以下操作：

import org.elasticsearch.spark._
val rdd = sc.esRDD("my-index")

如果我这样做rdd.getClass，我会得到以下结果：

res9: Class[_ <: org.apache.spark.rdd.RDD[(String, scala.collection.Map[String,AnyRef])]] = class org.elasticsearch.spark.rdd.ScalaEsRDD

我想这意味着我有一个带有字符串（文档 ID，我不需要它）和地图（JSON 本身）的元组列表。

现在我想用 JSON 提取 Map，创建一个模式并将数据以 Avro 格式放入 HDFS。

我尝试通过 val rdd_res = rdd.take(10) 获取仅 10 条记录用于测试目的，而 rdd_res.getClass 我得到了

res11: Class[_ <: Array[(String, scala.collection.Map[String,AnyRef])]] = class [Lscala.Tuple2

由于我对 Spark 和 Scala 缺乏经验，所以我很困惑。如何从数组的每个值中提取地图？接下来，如何将结果作为 Avro 保存到 HDFS？

提前感谢您的宝贵时间。

【问题讨论】：

欢迎部分解决方案或建议！

标签： scala apache-spark rdd

【解决方案1】：

我已经解决了我自己的问题，我把我发现的东西放在这里以供将来参考。对于读取我使用的数据：

val sc = new SparkContext()
val sql = new SQLContext(sc)

var df = sql.read.format("es").load("my-index-*/log")

这使您甚至可以从多个索引中获取数据（请注意*）。如果您需要在 spark-shell 中运行此代码，则可以省略第一个命令，如果您需要在编译 jar（使用 SBT 或 Maven）后在 spark-submit 中运行此代码，则应包含它。

读取数据后，您可以对其进行处理（例如，可以将列重命名为var df = df.withColumnRenamed("@timestamp", "ts")）。

处理完数据后，您可以使用出色的 Databrick spark-avro 库将它们导出为 Avro。您只需要使用 SBT 编译它，如果需要，在配置代理之后，在项目的根目录上运行 $ sbt package。然后在target/scala 路径下你可以找到你的jar。

使用此库，您可以按以下方式导出 Avro：

df.write
    .partitionBy("column")
    .mode("append")
    .format("com.databricks.spark.avro")
    .save("my/hdfs/path")

就是这样！如果不需要分区，可以省略partitionBy 函数，并且可以使用与append 不同的模式来保存数据。

我希望这个答案可以帮助到你。

更新

我忘了告诉你需要的导入。

import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.DataFrame
import org.elasticsearch.spark.sql._
import com.databricks.spark.avro._

【讨论】：