Не удалось загрузить DF в elasticsearch

Я читаю файл паркета в 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?


apache-spark apache-spark-sql apache-spark-2.0
person DarKnight    schedule 17.01.2019    source источник

Ответы (1)


arrow_upward
0
arrow_downward

Я смог отправить его, преобразовав их в строку. 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")
person DarKnight    schedule 28.03.2019
comment
Конечно, ES умен, но лучше, если вы создадите свое отображение заранее. Вы можете добавить пользовательский тип данных и изменить анализатор в соответствии с вашими потребностями (полнотекстовый, нечеткий...) - person Ragnar; 18.09.2019