限制从Apache Spark到ES的写入

文章/答案/技术大牛

发布

1回答

apache-spark、elasticsearch、throttling

在我们的项目中，我们使用Apache Spark来写入ES。我们运行多个并行写入ES的spark作业。我们处理的数据量太大，导致写入吞吐量高达约5K写入/秒。我们希望限制ES写入，以便将其降低到500到1000写入/秒的范围。我们遇到过像es</e

浏览 15提问于2020-07-20得票数 1

2回答

将数据从Apache* spark中的JavaDStream<String>写入到elasticsearch*

elasticsearch、apache-spark、apache-kafka

我正在编写程序来处理从Apache kafka到elasticsearch的数据。为此，我使用Apache Spark。我已经通过许多链接，但无法找到示例，以写入数据从JavaDStream在Apache spark到elasticsearch。import org.apache

浏览 2提问于2016-08-26得票数 0

1回答

使用Spark的Elasticsearch编写

apache-spark、elasticsearch

我正在以RDD的形式在Spark中创建一个文档集合，并使用来自Elasticsearch的Spark读写库。创建集合的群集很大，因此当它向ES写入时，我会得到下面指示ES的错误，这并不让我感到意外。在Spark中，报告的作业已成功完成。这些错误是否意味着某些数据没有写入索引？以下是许

浏览 3提问于2017-03-20得票数 0

1回答

如何用Spark写到远程弹性搜索节点？

apache-spark、hadoop、elasticsearch

我有一个代码：import org.apache.spark.SparkContextimport org.elasticsearch.spark._conf.set("es.index.auto.create",&qu

浏览 0提问于2019-06-11得票数 0

3回答

将Spark* 2.X连接到ElasticSearch 2.X*

hadoop、elasticsearch、apache-spark、elasticsearch-2.0

我正在使用Spark和ElasticSearch，但是我不知道如何在ElasticSearch 2.x中使用Spark 2.X。ElasticSearch Spark libs只支持ES2.x的Spark1.6，支持ES5.alpha的Spark2。你用什么将Spark连接到ElasticSearch？

浏览 0提问于2016-09-20得票数 1

1回答

为什么查询抛出ClassCastException "SerializedOffset不能通过MQTT源转换到ClassCastException？“

spark-structured-streaming、apache-bahir

火花结构化流代码如下所示，基本上我只是从MQTT队列读取数据并写入ElasticSearch索引。") .option("es.nodes", "localhost") .start() .awaitTermination

浏览 1提问于2018-12-05得票数 1

回答已采纳

1回答

如何使用Spark在ElasticSearch中保存JSON文件？

scala、apache-spark、elasticsearch

这是我的代码：import org.apache.spark.sql.SQLContextimport org.apache.spark.SparkConf def main(args: Array[String]): Unit = {val conf = ne

浏览 4提问于2018-07-19得票数 2

1回答

Spark structured streaming Elasticsearch集成问题

apache-spark、elasticsearch、streaming

我正在编写一个Spark结构的流媒体应用程序，其中使用Spark处理的数据需要沉没到弹性搜索。这是我的开发环境。() 我尝试了以下两种方法将DataSet中的数据汇聚到ES中。("es").start("rt_request/doc")；在这两种情况下，我都得到以下错误：原因: java.lang.UnsupportedOperationException:数据源es</em

浏览 27提问于2019-06-12得票数 0

回答已采纳

1回答

无法从主机的传输池获取传输

apache-spark、elasticsearch、ibm-cloud、compose-db、analytics-engine

我正在尝试从IBM Analytics Engine上的Spark Structured Streaming写入IBM Compose Elasticsearch接收器。我的spark代码： .writeStream .format("org.elasticsearch.spark.sql", "admin") .option(&qu

浏览 12提问于2018-09-03得票数 1

回答已采纳

3回答

不安全模式下的Elasticsearch火花放电连接

apache-spark、elasticsearch、curl、pyspark、elasticsearch-hadoop

我的最终目标是将数据从hdfs插入到elasticsearch，但我面临的问题是连接性curl -u username -X GETv' --insecure 但是当涉及到与火花的联系时，我无法做到这一点。我插入数据的命令是df.write.mode("append").format('org.elasticsearch.spark</

浏览 5提问于2020-08-10得票数 0

回答已采纳

5回答

Elasticsearch + Spark:用自定义文档_id编写json

scala、apache-spark、elasticsearch、elasticsearch-hadoop

at org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:105)当我从配置中删除es.mapping.id和es.mapp

浏览 8提问于2017-12-19得票数 1

1回答

使用ES Hadoop连接器在Elastic Search中保存JavaRDD

java、hadoop、apache-spark、elasticsearch、spark-dataframe

目前正在处理一个转换项目，我需要将数据从Oracle提供给elastic search。所以我的工作是这样的2.Java Spark - Dataframe Joins then saving them into elastic search repo's{ }Field 5:{ -- Maps

浏览 0提问于2017-05-12得票数 0

1回答

从spark查询弹性搜索大索引

scala、apache-spark、elasticsearch、apache-spark-sql

我想要的是阅读整个索引从星火转换为拼花格式。:190) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql我通过手动过滤(按压)数据来修正这个问题，从而减少了向Elasticsearch请求的数据量。我使用时间戳来限制查询的响应。为了能够读取整个索引，我不得不多次查询Elasticsearch。

浏览 0提问于2017-11-23得票数 1

1回答

Spark-Streaming挂起，kafka最早开始偏移(Kafka 2，spark 2.4.3)

apache-spark、apache-kafka、kafka-consumer-api、spark-structured-streaming

我对Spark-Streaming和Kafka有意见。在运行示例程序从Kafka主题消费并将微批结果输出到终端时，当我设置选项时，我的作业似乎挂起了： df.option("startingOffsets", "earliest") 从最新的偏移量开始工作很好，结果随着每个微批次的流过而打印到终端。我在想，也许这是一个资源问题--我正在尝试从一个有相当多数据的主题中阅读。但是，我似乎没有内存/cpu问题(使用本地*集群运

浏览 21提问于2019-09-18得票数 4

回答已采纳

2回答

EsHadoopIllegalArgumentException:无法检测ES版本Spark示例

scala、apache-spark、elasticsearch、spark-streaming

我正在尝试运行简单的数据写入ElasticSearch示例。但是，我一直收到这样的错误： &qu

浏览 0提问于2018-05-08得票数 1

回答已采纳

1回答

用火花2.x连接到elasticsearch 2.4.4

java、apache-spark、elasticsearch

从官方的DOC我们可以看到： .set("spark.serializer", "org.apache.spark.serializer.KryoSerializ

浏览 3提问于2017-07-26得票数 0

回答已采纳

1回答

Spark流数据帧对elasticsearch公开

apache-spark、elasticsearch、spark-streaming

SparkSession配置： .config("es.index.auto.create", "true") .config$.withNewExecutionId(Lorg/apache/spark/sql/SparkSession;Lorg/apache/spark/sql/ex

浏览 0提问于2021-02-22得票数 0

1回答

如何从Elasticsearch读取数据到Spark？

java、python、scala、elasticsearch、apache-spark

我正试着用python把数据从ElasticSearch读到Apache Spark。$ ."} rdd = sc.newAPIHadoopRDD("org.elasticsearch.hadoop.mr.EsInputFormat", "org.apache.hadoop.io.NullWritable", "org.elasticsearch.hadoop.mr.LinkedM

浏览 1提问于2016-03-14得票数 1

回答已采纳

1回答

如何在启用X-Pack的情况下设置Elasticsearch Structured Streaming？

apache-spark、spark-structured-streaming、elasticsearch-hadoop

我正在尝试使用安装了x-pack的Elasticsearch (ES) 6.1.1 Hadoop来使用Spark Structured Streaming 2.2.1写入数据。这是我的代码(索引已经存在于elastic中)： .readStreamval advancedQuery = exceptionsrunningQuery.awaitTerminationorg.elast

浏览 4提问于2018-01-17得票数 0

1回答

解耦不可串行化对象以避免星火中的序列化错误

scala、serialization、apache-spark

下面的类包含试图从Elasticsearch读取并打印返回的文档的主要函数： sparkConf.set("spark.kryoserializer.buffer","256") val es = new EsContext(sparkConf)at org.apache<

浏览 3提问于2016-03-10得票数 1

回答已采纳

点击加载更多