r - 如何使用SparkR读写ElasticSearch？

Question

初学者SparkR和ElasticSearch在这里提问!

如何将带有多个节点的sparkR数据帧或RDD写入ElasticSearch？

存在一个specific R package for elastic，但是没有提到hadoop或分布式数据帧。当我尝试使用它时，出现以下错误:

install.packages("elastic", repos = "http://cran.us.r-project.org")
library(elastic)
df <- read.json('/hadoop/file/location')
connect(es_port = 9200, es_host = 'https://hostname.dev.company.com', es_user = 'username', es_pwd = 'password')
docs_bulk(df)

Error: no 'docs_bulk' method for class SparkDataFrame

如果这是pyspark，我将使用 rdd.saveAsNewAPIHadoopFile() function as shown here，但在谷歌搜索中找不到sparkR的任何相关信息。 ElasticSearch也有 good documentation, but only for Scala and Java

我敢肯定我很想念一些东西；任何指导表示赞赏!

Answer 1

要将SparkR session 连接到Elasticsearch，您需要使连接器jar和ES配置可用于SparkR session 。

1:指定 jar (在elasticsearch文档中查找所需的版本；以下版本适用于spark 2.x，scala 2.11和ES 6.8.0)

sparkPackages <- "org.elasticsearch:elasticsearch-spark-20_2.11:6.8.0"

2:在SparkConfig中指定集群配置。您也可以在此处添加其他Elasticsearch配置(当然，还可以添加其他spark配置)

sparkConfig <- list(es.nodes = "your_comma-separated_es_nodes",
                    es.port = "9200")

启动sparkR session

sparkR.session(master="your_spark_master", 
               sparkPackages=sparkPackages, 
               sparkConfig=sparkConfig)

做一些魔术操作，从而生成一个您想保存到ES的sparkDataframe

将您的数据框写入ES:

write.df(yourSparkDF, source="org.elasticsearch.spark.sql",
                 path= "your_ES_index_path"
         )