1. spark 读取 ES
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark.rdd.Esspark
object esReadToHdfs {
def main(args: Array[String]): Unit = {
val spark = SparkSession.builder().appName("es_read").getorCreate()
val sc = spark.sparkContext
val options = Map(
"es.index.auto.create" -> "true",
"es.nodes.wan.only" -> "true",
"es.nodes" -> "29.29.29.29:9200,29.29.29.29:9200",
"es.port" -> "9200",
"es.mapping.id" -> "id"
)
// 返回 RDD[(String, String]]
// 元组：第一个：esmapping.id、第二个 json 字符串
val resultRDD = Esspark.esJsonRDD(sc, options).map(x => x._2)
// // 返回 RDD[(String, Map[String, AnyDef]]
// val resultRDD = Esspark.esRDD(sc, options)
}
}
2. spark 写入 ES
参考文章
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
