如何在PySpark中用TICKETID+CUSTID复合键作为ES映射ID实现Upsert

Problem Background #

You’re currently using TICKETID as the document ID for Elasticsearch upserts, but this field isn’t unique on its own—only the combination of TICKETID and CUSTID forms a unique key in your DB2 source table. Since Elasticsearch’s es.mapping.id configuration only accepts a single field name, you’ll need to adjust your PySpark code to create a composite identifier first.

Solution Steps #

1. Create a composite key field in your DataFrame by concatenating TICKETID and CUSTID (use a separator like a hyphen to avoid accidental collisions where overlapping values might create duplicate IDs).
2. Update the es.mapping.id option to point to this new composite field.

Modified Code Example #

# Import necessary function if not already imported
from pyspark.sql.functions import concat_ws, col

# First, create the composite unique key field
db2_df_with_composite_id = db2_df.withColumn(
    "COMPOSITE_ID",
    concat_ws("-", col("TICKETID"), col("CUSTID"))
)

# Write the data into Elasticsearch with the composite ID
db2_df_with_composite_id.write \
 .format("org.elasticsearch.spark.sql") \
 .option("checkpointLocation", "/soumendra/chkpnt_files/1") \
 .option("es.nodes.wan.only", "true") \
 .option("es.nodes", "ES_IP:PORT") \
 .option("es.write.operation", "upsert") \
 .option("es.mapping.id", "COMPOSITE_ID") \
 .mode("append") \
 .save("my_index")

Key Notes #

• Using concat_ws (concat with separator) ensures that combinations like TICKETID=123 + CUSTID=45 and TICKETID=12 + CUSTID=345 don’t produce the same ID (they become 123-45 vs 12-345).
• Numeric fields will be automatically cast to strings for concatenation, but you can add cast(StringType()) explicitly if you want to enforce string formatting.
• If you don’t want the COMPOSITE_ID field to appear in your Elasticsearch index, you can drop it after writing, or use the es.mapping.exclude option to exclude it from the document mapping.

内容的提问来源于stack exchange，提问作者Soumendra