问题标签 [elasticsearch-hadoop]

我正在使用 Databrics 来运行我的 Spark 应用程序，并且我正在尝试使用它elasticsearch-hadoop来建立与 Elasticsearch 的连接。

在我的 Databricks VPC 和我的 Elasticsearch VPC 之间配置对等连接后，我终于可以让两者相互交谈。它看起来像这样：

这向我表明它们是相连的，太棒了！

但最终的测试还是失败了。这是我尝试运行的内容：

df 就是这样：

但我得到的错误是有害的：

起初我认为这与版本错误有关，但我正在努力寻找不匹配的地方。

我在跑：

• AWS 弹性搜索 - 7.1

• Databricks - 6.1（包括 Apache Spark 2.4.4、Scala 2.11）

• elasticsearch-hadoop - elasticsearch_spark_20_2_11_7_5_2.jar

我错过了什么？

我在将数据从 Elasticsearch 读取到 Spark 集群时遇到问题（我使用的是 Zeppelin 环境，因此所有连接设置都在 Zeppelin 解释器设置中配置）。

首先，我尝试使用 PySpark 阅读它：

不幸的是，在这种情况下，我面临许多映射问题。因为我在数据集中有很多包含点的字段，我决定让 Scala 尝试读取数据（以便稍后在 PySpark 中处理它）：

但是，即使使用 Scala，我也只能检索少量记录，例如

出于某种原因，collect() 不起作用：

错误是：

我也尝试过转换为 DF，但出现错误：

您对如何处理有任何想法吗？

我正在尝试从 s3 读取并写入 Elasticsearch，在 spark master 机器上使用 jupyter install

我有这个配置：

使用此配置，当尝试使用此配置从 s3 读取时，我可以访问 ES 而不是 S3，我收到此错误：

Py4JJavaError：调用 z:org.apache.spark.api.python.PythonRDD.collectAndServe 时出错。：java.lang.RuntimeException：java.lang.ClassNotFoundException：类org.apache.hadoop.fs.s3native.NativeS3FileSystem找不到

当禁用 sc_conf.set('spark.packages'.. 和 sc_conf.set('spark.jars', .. 并启用 #os.environ['PYSPARK_SUBMIT_ARGS'] 时，它确实可以访问 s3 但不能访问 ES

我想念什么？

谢谢亚尼夫

使用Elasticsearch-hadoop连接器，是否可以在 upsert 插入时使用scripted_upsert为 true ？

我正在使用es.update.script.inline配置，但我找不到任何方法来使用 script_upsert 为 true 并清空 upsert 的内容

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

结果是：

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

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

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

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

我正在使用Spark (2.2.0)我ElasticSeach Hadoop (7.6.0) 的 Spark Job 的目的是处理 parquet 文件中的记录，并将其附加到 ElasticSearch 中已经存在的文档中。由于 ElasticSearch 不支持更新，因此获取记录和更新记录的部分由作业处理。

我大约20 million records在索引中。在任何时候，我都不需要所有记录，因此我使用过滤器下推来仅获取所需数量的文档。

出于性能原因，您可以下推的最大记录词数是65536. 我提出了100K但没有进一步移动它，因为获取的平均记录数介于2-3 million.

因此，目标是创建数据帧，100K每个请求获取记录并使用union

我的部分代码如下

使用上面的代码，Spark UI 在函数完成后卡住了，没有启动任何任务collect()，直到我收到 OOM 错误。

为了尝试使用 RDD 进行并行化，我尝试了以下方法：

抛出一个NullPointerException

我理解第二种方法的问题，因为DataFrame和RDD是抽象的驱动程序概念，因此执行程序无法对它们进行操作。

但是在尝试了所有这些之后，我没有想法，还有什么可以尝试的。如果有人能指出我正确的方向，我将不胜感激。

谢谢 ！！

更新：更新了代码片段以更密切地关注问题。

我的最终目标是将数据从 hdfs 插入到 elasticsearch 但我面临的问题是连接性

我可以使用以下 curl 命令连接到我的 elasticsearch 节点

curl -u username -X GET https://xx.xxx.xx.xxx:9200/_cat/indices?v' --insecure

但是当谈到与火花的连接时，我无法这样做。我插入数据的命令是 df.write.mode("append").format('org.elasticsearch.spark.sql').option("es.net.http.auth.user", "username").option("es.net.http.auth.pass", "password").option("es.index.auto.create","true").option('es.nodes', 'https://xx.xxx.xx.xxx').option('es.port','9200').save('my-index/my-doctype')

我得到的错误是

在这里，pyspark 相当于 curl --insecure

谢谢

使用带有 elasticsearch-hadoop 库的 Spark 读取 Elasticsearch 记录时，时间戳无效。我正在使用以下 Spark 代码进行记录读取：

这导致以下结果：

但我期待当前年份的时间戳，例如2021-01-19 16:04:27.228。在弹性中，start_time字段具有以毫秒为单位的 unixtime 格式 ->start_time": 1611314773.641

错误：

java.lang.ArrayIndexOutOfBoundsException：在 org.apache.Hadoop.mapreduce.Mapper.run(Mapper .java:146) 在 org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:799) 在 org.apache.hadoop.mapred.MapTask.run(MapTask.java:347) 在 org.apache.hadoop。 mapred.YarnChild$2.run(YarnChild.java:174) at java.base/java.security.AccessController.doPrivileged(AccessController.java:691) at java.base/javax.security.auth.Subject.doAs(Subject.java :425) 在 org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1762) 在 org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:168)

在使用 elasticsearch-hadoop 库读取具有空属性的弹性搜索索引时，出现异常

github 中存在相同的开放缺陷，并具有重现它的步骤：https ://github.com/elastic/elasticsearch-hadoop/issues/1635

Spark：3.1.1
Elasticsearch-Hadoop：elasticsearch-spark-30_2.12-7.12.0
Elasticsearch：2.3.4