1.建立spark與Elasticsearch的連線

為了對Elasticsearch進行讀寫操作，需要新增Elasticsearch的依賴包，其中，新增依賴包（org.elasticsearch_elasticsearch-spark-20_2.11-6.8.7.jar）有下面的三種方式：

1）將依賴包直接放在安裝spark目錄下面的jars目錄下，即可；

2) 在提交任務時，利用spark submit --jars 的方式

3）在建立spark物件時，新增依賴，如下圖所示

spark = SparkSession \
    .builder \
    .appName('es connection') \
    .config('spark.jars.packages', "org.elasticsearch_elasticsearch-spark-20_2.11-6.8.7") \
    .getOrCreate()

2.spark 讀取Elasticsearch的資料

df3 = spark.read \
    .format("org.elasticsearch.spark.sql") \
    .option("es.nodes", '節點') \
    .option('es.port', '埠') \
    .option("es.resource", '索引/索引型別') \
    .option('es.query', '?q=*') \
    .option('es.nodes.wan.only','true') \
    .option("es.nodes.discovery", "false") \
    .option("es.index.auto.create", "true") \
    .option("es.write.ignore_exception", "true") \
    .option("es.read.ignore_exception","true") \
    .load()

3.spark 寫入elasticsearch

df.write.format('org.elasticsearch.spark.sql') \
        .option('es.nodes', '節點') \
        .option('es.port', '9200') \
        .option('es.nodes.wan.only', 'true') \
        .option("es.nodes.discovery", "false") \
        .option('es.resource', '索引/索引型別') \
        .save(mode='append')

備註：

當spark讀寫elasticsearch的過程中，elasticsearch包含Array型別的欄位，就會出現下面錯誤：

無法將List型別資料寫入到es, 或者從es讀出list型別資料

解決方案：

在option 中新增一個es.read.field.as.array.include屬性，value為list Schema的欄位名