Spark 结构化流 Elasticsearch 集成问题

Question

我正在编写一个 Spark 结构化流应用程序，其中使用 Spark 处理的数据需要下沉到弹性搜索。这是我的开发环境。 Hadoop 2.6.0-cdh5.16.1 星火版本 2.3.0.cloudera4 弹性搜索 6.8.0

我运行火花-shell作为

spark2-shell --jars /tmp/elasticsearch-hadoop-2.3.2/dist/elasticsearch-hadoop-2.3.2.jar

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.sql.functions._
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{StructType, StructField, StringType, IntegerType, TimestampType};
import java.util.Calendar
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.sql
import sys.process._

val checkPointDir =  "/tmp/rt/checkpoint/"

val spark = SparkSession.builder
.config("fs.s3n.impl", "org.apache.hadoop.fs.s3native.NativeS3FileSystem")
.config("fs.s3n.awsAccessKeyId","aaabbb")
.config("fs.s3n.awsSecretAccessKey","aaabbbccc")
.config("spark.sql.streaming.checkpointLocation",s"$checkPointDir")
.config("es.index.auto.create", "true").getOrCreate()
import spark.implicits._

val requestSchema = new StructType().add("log_type", StringType).add("time_stamp", StringType).add("host_name", StringType).add("data_center", StringType).add("build", StringType).add("ip_trace", StringType).add("client_ip", StringType).add("protocol", StringType).add("latency", StringType).add("status", StringType).add("response_size", StringType).add("request_id", StringType).add("user_id", StringType).add("pageview_id", StringType).add("impression_id", StringType).add("source_impression_id", StringType).add("rnd", StringType).add("publisher_id", StringType).add("site_id", StringType).add("zone_id", StringType).add("slot_id", StringType).add("tile", StringType).add("content_id", StringType).add("post_id", StringType).add("postgroup_id", StringType).add("brand_id", StringType).add("provider_id", StringType).add("geo_country", StringType).add("geo_region", StringType).add("geo_city", StringType).add("geo_zip_code", StringType).add("geo_area_code", StringType).add("geo_dma_code", StringType).add("browser_group", StringType).add("page_url", StringType).add("document_referer", StringType).add("user_agent", StringType).add("cookies", StringType).add("kvs", StringType).add("notes", StringType).add("request", StringType)
val requestDF = spark.readStream.option("delimiter", "\t").format("com.databricks.spark.csv").schema(requestSchema).load("s3n://aa/logs/cc.com/r/year=" + Calendar.getInstance().get(Calendar.YEAR) + "/month=" + "%02d".format(Calendar.getInstance().get(Calendar.MONTH)+1) + "/day=" + "%02d".format(Calendar.getInstance().get(Calendar.DAY_OF_MONTH)) + "/hour=" + "%02d".format(Calendar.getInstance().get(Calendar.HOUR_OF_DAY)) + "/*.log")
requestDF.writeStream.format("org.elasticsearch.spark.sql").option("es.resource", "rt_request/doc").option("es.nodes", "localhost").outputMode("Append").start()

我尝试了以下两种方法将DataSet中的数据下沉到ES中。 1.ds.writeStream().format("org.elasticsearch.spark.sql").start("rt_request/doc"); 2.ds.writeStream().format("es").start("rt_request/doc"); 在这两种情况下，我都会收到以下错误：

原因： java.lang.UnsupportedOperationException: 数据源es不支持流式写入

java.lang.UnsupportedOperationException:数据源org.elasticsearch.spark.sql不支持流式写入在 org.apache.spark.sql.execution.datasources.DataSource.createSink(DataSource.scala:320) 在 org.apache.spark.sql.streaming.DataStreamWriter.start(DataStreamWriter.scala:293) ... 57 被省略

Answer 1

ES-hadoop 我用的jar版本是旧的elasticsearch-hadoop-2.3.2.jar。我们需要6个或以上。

现在我使用 elasticsearch-hadoop-6* 或更高版本的 jars 作为流媒体接收器。

我已经从https://artifacts.elastic.co/downloads/elasticsearch-hadoop/elasticsearch-hadoop-7.1.1.zip

下载了它

Spark 结构化流 Elasticsearch 集成问题

Spark structured streaming Elasticsearch integration issue

apache-spark

elasticsearch

streaming