问题标签 [elasticsearch-spark]

由于“未找到 id 的搜索上下文”，Spark 任务失败。我尝试了几个选项，例如

spark.es.input.max.docs.per.partition 250

spark.es.scroll.size 100

spark.es.batch.size.bytes 32mb

但任务仍然失败。我们正在使用： Spark 版本：2.3.1，Elasticsearh 集群版本：6.5.4，elasticsearch-spark 版本：6.5.4

任何帮助都将不胜感激。

我想根据我的 spark 转换在 Elasticsearch 中创建一个索引。我想知道使用 elasticsearch-spark 库的最佳方法是什么？亲切的问候

我正在从 Elasticsearch 索引中读取 PySpark SQL Dataframe，读取选项为es.read.metadata=True. 我想按元数据字段上的条件过滤数据，但得到一个空结果，尽管应该有结果。是否有可能得到实际结果？

当我在数据帧上使用时，我确实得到了结果limit，即使数字很大，甚至比数据帧的大小还要大。

此外，我在使用其他不_metadata相关的字段时确实得到了结果。

例如：

结果为空：

但是使用时limit：

结果不为空：

线程“主”java.lang.NoSuchMethodError 中的异常：org.elasticsearch.spark.sql.EsSparkSQL$.saveToEs

将文档从 Spark 应用程序插入 Elasticsearch 集群时出现上述错误。

我有一个火花数据框，其中有一列说“名称”。该名称在单个数据框中可能具有不同的值。当我使用 spark (scala) 将数据写入 elasticsearch 时，我想根据列“name”的值将数据写入不同的索引。

saveToEs 需要一个字符串，我正在寻找以下内容：

或类似的东西，我可以在写入期间分配值。

我有一张地图如下：

我的弹性数据有一个名为“名称”的字段。我想从 spark 查询弹性搜索并返回所有具有“a”、“b”、“c”作为“name”字段值的记录。

我不想硬编码弹性查询中的名称。

有人可以帮我构建一个动态查询，它接受字符串列表。

我正在尝试将包含结构列的数据框写入 Elasticsearch：

结果是：

这确实有效，但 JSON 被转义，因此相应的详细信息字段在 Kibana 中不可点击：

我尝试提供.option("es.input.json","true")，但得到一个异常：

相反，如果我尝试在不转换为 JSON 的情况下写入数据，即从原始代码中删除to_json( ，我会得到另一个异常：

所以问题是如何将带有嵌套 JSON 列的 PySpark 数据框写入 Elasticsearch，这样 JSON 就不会被转义？

我正在做一个将数据从 Elastic Search 加载到 HDFS 的 Spark 程序，但在初始化 SparkContext 时出现错误。运行作业时出错。错误是在制作火花会话期间。

Hadoop：3.2.1

火花：2.4.4

Elasticsearch Spark（适用于 Spark 2.X）» 7.5.1

电子病历：6.0.0

代码：

错误日志：

有人可以帮忙吗？谢谢。

我正在将使用elasticsearch-spark 7.6.0版本的现有应用程序迁移到最新的 ie elasticsearch-spark-30_2.12:7.15.0。我正在加载带有日期映射的 ES 数据，如下所示

但是我收到错误

我无法弄清楚虽然值很长并且打印数据集dataset.show()显示有效日期，但为什么无法写入 ES？是否需要为具有最新 ES 的日期字段传递任何新配置？任何帮助表示赞赏。

我想写一个从 Kafka 到 Elasticsearch 的 Spark Streaming Job。在这里，我想在从 Kafka 读取模式时动态检测模式。

你能帮我这样做吗？

我知道，这可以通过下一行在 Spark 批处理中完成。

val schema = spark.read.json(dfKafkaPayload.select("value").as[String]).schema

但是在通过 Spark Streaming Job 执行相同的操作时，我们无法执行上述操作，因为流式处理只能在 Action 上进行。

请告诉我。