无法将DF摄取到elasticsearch

Question

我正在阅读spark-scala中的镶木地板文件并进行计算和过滤。我想将结果数据帧摄取到elasticsearch。我试过跟随https://www.elastic.co/guide/en/elasticsearch/hadoop/current/spark.html#spark-sql，但无法使它工作。

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession, SQLContext}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SQLContext._
import org.elasticsearch.spark._

val spark = SparkSession.builder.appName("test dumper").config("es.index.auto.create", "true")
  .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
  .config("es.nodes", "<ip>").config("es.port", "<port>").getOrCreate()

val sc = spark.sparkContext
sc.hadoopConfiguration.set("mapreduce.fileoutputcommitter.algorithm.version", "2")
....... // Doing some filtering

df.rdd.saveToEs("testing/2019")

这会引发错误：

org.elasticsearch.hadoop.serialization.EsHadoopSerializationException: org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Spark SQL types are not handled through basic RDD saveToEs() calls; typically this is a mistake(as the SQL schema will be ignored). Use 'org.elasticsearch.spark.sql' package instead
at org.elasticsearch.hadoop.serialization.bulk.BulkEntryWriter.writeBulkEntry(BulkEntryWriter.java:136)
at org.elasticsearch.hadoop.rest.RestRepository.writeToIndex(RestRepository.java:170)
at org.elasticsearch.spark.rdd.EsRDDWriter.write(EsRDDWriter.scala:67)
at org.elasticsearch.spark.rdd.EsSpark$$anonfun$doSaveToEs$1.apply(EsSpark.scala:107)
at org.elasticsearch.spark.rdd.EsSpark$$anonfun$doSaveToEs$1.apply(EsSpark.scala:107)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Caused by: org.elasticsearch.hadoop.EsHadoopIllegalArgumentException: Spark SQL types are not handled through basic RDD saveToEs() calls; typically this is a mistake(as the SQL schema will be ignored). Use 'org.elasticsearch.spark.sql' package instead
at org.elasticsearch.spark.serialization.ScalaValueWriter.doWriteScala(ScalaValueWriter.scala:124)
at org.elasticsearch.spark.serialization.ScalaValueWriter.write(ScalaValueWriter.scala:46)
at org.elasticsearch.hadoop.serialization.builder.ContentBuilder.value(ContentBuilder.java:53)
at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk.doWriteObject(TemplatedBulk.java:71)
at org.elasticsearch.hadoop.serialization.bulk.TemplatedBulk.write(TemplatedBulk.java:58)
at org.elasticsearch.hadoop.serialization.bulk.BulkEntryWriter.writeBulkEntry(BulkEntryWriter.java:68)
... 10 more

有没有办法直接将数据框摄取到elasticsearch？

Answer 1

我能够通过将它们转换为字符串来发送它。 ES巧妙地解释数据类型。

df.rdd.map(row => {
      var m = Map[String, Any]()
      (0 until len).foreach(i => {
        m += (schema.fields(i).name -> row.getAs[String](i))
      })
      m
    }).saveToEs("path")