Python如何把Spark数据写入ElasticSearch

Python可以使用ElasticSearch的Python客户端库（Elasticsearch-py）来将Spark数据写入Elasticsearch。下面我们来讲解一下具体的步骤。

1. 安装 Elasticsearch-py

pip install elasticsearch

2. 在Spark中创建DataFrame

首先需要在Spark中加载要写入Elasticsearch的数据集并将其转换为DataFrame格式。下面是一个示例代码片段：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 创建SparkSession
spark = SparkSession.builder.appName("Write to ElasticSearch").getOrCreate()

# 定义DataFrame的schema
schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

# 创建DataFrame
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
df = spark.createDataFrame(data, schema)

# 展示DataFrame
df.show()

3. 使用 Elasticsearch-py 库将DataFrame写入到Elasticsearch

在将DataFrame写入到Elasticsearch之前，需要将DataFrame中的数据转换为Python字典，然后使用Elasticsearch-py库将字典写入到Elasticsearch索引中。下面是一个示例代码片段：

from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

# 连接Elasticsearch
es = Elasticsearch(['localhost'], port=9200)

# DataFrame中的数据转换为Python字典
data_dict = df.rdd.map(lambda x: x.asDict()).collect()

# 写入Elasticsearch
bulk_data = []
for d in data_dict:
    op_dict = {
        "index": {
            "_index": "my_index",
            "_type": "my_type"
        }
    }
    op_dict.update(d)
    bulk_data.append(op_dict)

bulk(es, bulk_data)

上面的代码将Python字典的数据逐一加入列表bulk_data中，然后用bulk()函数进行批量插入。

示例说明

示例1：从CSV文件中加载数据

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder.appName("Write to ElasticSearch").getOrCreate()

# 读取CSV文件并将其转换为DataFrame
df = spark.read\
    .option("header", "true")\
    .option("inferSchema", "true")\
    .csv("file:///path/to/file.csv")

# 写入Elasticsearch
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

es = Elasticsearch(['localhost'], port=9200)

data_dict = df.rdd.map(lambda x: x.asDict()).collect()

bulk_data = []
for d in data_dict:
    op_dict = {
        "index": {
            "_index": "my_index",
            "_type": "my_type"
        }
    }
    op_dict.update(d)
    bulk_data.append(op_dict)

bulk(es, bulk_data)

如果数据在CSV文件中存在，可以直接从文件中加载数据，读取方式可以根据情况自行选择。

示例2：从Hive表中加载数据

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder.appName("Write to ElasticSearch").getOrCreate()

# 从Hive表中加载数据并将其转换为DataFrame
df = spark.sql("SELECT * FROM my_hive_table")

# 写入Elasticsearch
from elasticsearch import Elasticsearch
from elasticsearch.helpers import bulk

es = Elasticsearch(['localhost'], port=9200)

data_dict = df.rdd.map(lambda x: x.asDict()).collect()

bulk_data = []
for d in data_dict:
    op_dict = {
        "index": {
            "_index": "my_index",
            "_type": "my_type"
        }
    }
    op_dict.update(d)
    bulk_data.append(op_dict)

bulk(es, bulk_data)

如果数据存储在Hive中的表中，可以通过SQL语句将其加载到DataFrame中，代码类似于上述示例。

至此，Python通过Elasticsearch-py将Spark数据写入Elasticsearch的完整攻略就讲解完毕了。