问题标签 [elasticsearch-hadoop]

我有从 Cassandra 读取数据、处理/转换/过滤数据并将结果写入 Elasticsearch 的 spark 作业。我使用 docker 进行集成测试，但在从 spark 到 Elasticsearch 的编写过程中遇到了麻烦。

依赖项：

在我的单元测试中，我可以使用 TransportClient 连接到 elasticsearch 来设置我的模板和索引

又名。这有效

但是，当我尝试运行时

我收到这个堆栈跟踪

我可以使用 'docker network inspect bridge 验证它正在尝试连接到正确的 IP 地址。

我在 macbook/osx 上本地运行所有内容。我不知道为什么我可以使用 TransportClient 并通过我的浏览器连接到 docker 容器，但是函数 EsSpark.saveToES(...) 总是失败。

当我执行具有以下功能的 Scala 应用程序时引发异常myRDD.saveToEs（我还尝试了数据帧中的 saveToEs）。我的 ES 版本是2.3.5. 我正在使用 Spark 1.5.0，所以也许有一种SparkContext我不知道的方法来配置它。

堆栈跟踪如下 -

命名的字段foo_eff_dt确实有值，在某些情况下没有（即空）。我不确定这是否导致异常。

我的 scala 代码片段如下所示：

请帮助/指导我解决这个问题。

TIA。

我正在尝试重新索引 ElasticSearch，我使用了 Scan 和 Bulk API，但它非常慢，我怎样才能并行处理以使其更快。我的python代码如下：

我应该进行扫描多个进程还是应该进行批量多个进程。我一直在徘徊 ElasticSearch-Hadoop 如何实现这一点。我的索引有 10 个节点和 20 个分片。

是否可以使用一些“自定义标准”/模糊匹配（例如数字或日期的范围/间隔以及字符串的各种“距离方法”（例如 Levenshtein）来连接两个(Pair)RDDs（或Datasets/ s）（在多个字段上）？DataFrame

对于在 an 中“分组”RDD以获得 a PairRDD，可以实现 a PairFunction，但是在 JOINing 两个RDDs/数据集时似乎不可能进行类似的操作？我在想类似的事情：

我正在考虑在其中实现自定义逻辑hashCode()，equals()但我不确定如何使“相似”数据在同一个存储桶中结束。我也一直在研究，RDD.cogroup()但还没有弄清楚如何使用它来实现这一点。

我刚遇到elasticsearc-hadoop。有谁知道该库是否可以用来做这样的事情？

我正在使用 Apache Spark 2.0.0。我正在用 Java 实现，但 Scala 中的答案也会非常有帮助。

PS。这是我的第一个 Stackoverflow 问题，如果我犯了一些新手错误，请多多包涵:)。

我不知道如何使用来自 spark 的 python 将数据帧写入 elasticsearch。我按照这里的步骤进行操作。

这是我的代码：

上面的代码给出

引起：net.razorvine.pickle.PickleException：ClassDict 构造的预期零参数（对于 pyspark.sql.types._create_row）

我还从以下位置启动脚本： spark-submit --master spark://aggregator:7077 --jars ../jars/elasticsearch-hadoop-2.4.0/dist/elasticsearch-hadoop-2.4.0.jar /vagrant/scripts/aggregation.py以确保elasticsearch-hadoop已加载

我想知道是否有人有使用Elasticsearch for Hadoop library配置 Kubernetes 集群的经验。尝试从 spark 写入到 elasticsearch 时，我遇到了节点发现超时的问题。由于 ES的 elasticsearch-cloud-kubernetes插件，我启动并运行了 Elasticsearch ，它处理发现，但我不确定如何最好地配置 elasticsearch-hadoop 以了解 kubernetes 集群中的节点（pod）。我尝试设置spark.es.nodes为 es-client 服务，但这似乎不起作用。我也知道我可以启用es.nodes.wan.only，但如文档中所述，这将严重影响性能，这违背了让它们在同一集群上运行的目的。任何帮助，将不胜感激。

我有两个独立的 Elasticsearch 集群，我想将数据从第一个集群重新索引到第二个集群，但我发现我只能在 SparkContext 配置中设置一个 Elasticsearch 集群，例如：

那么如何在同一应用程序内的 Spark 中使用弹性搜索 hadoop 在两个 Elasticsearch 集群之间移动数据？

我正在将一些 Spark 1.6 代码更新为 2.0.1，并且在使用 map 时遇到了一些问题。

我在 SO 问题上看到了其他问题，例如encoder-error-while-trying-to-map-dataframe-row-to-updated-row但我无法让这些技术发挥作用，对于下面的这种情况来说它们似乎很荒谬。

在使用 Elasticsearch-Spark 连接器编写在 Apache Spark 1.6 上运行的应用程序时，（https://www.elastic.co/guide/en/elasticsearch/hadoop/5.0/install.html#_minimalistic_binaries）上的文档说要使用 maven artefact <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch-spark_2.10</artifactId> <version>5.0.0</version> </dependency>

但是，maven 失败并显示无法找到指定的人工制品的错误消息。在通过 maven central 搜索后，我发现了替代版本：

对于 Spark 1.3+

对于 Spark 2.*

我可能通读了 es-hadoop / es-spark 项目使用批量索引的 Stackoverflow。如果确实如此，则默认批量大小是根据 BulkProcessor(5Mb)。是否有任何配置可以改变这一点。

我JavaEsSparkSQL.saveToEs(dataset,index)在我的代码中使用，我想知道可用于调整性​​能的可用配置是什么。这是否也与数据集的分区有关。