如何将Avro转换为SQL批量Upsert语句？求简洁实现方案

Hey Muhammad, let's break down straightforward, no-custom-script solutions for both of your workflow ideas to sync that external RDBMS data to your local DB via Avro:

思路1: Avro直接转批量UPSERT语句（全原生处理器实现） #

This is the most efficient path, and NiFi has exactly the tools you need to skip manual conversion steps:

1. Pull data from external RDBMS: Use ExecuteSQL (or GenerateTableFetch + FetchSQL for large datasets) to fetch your data. By default, ExecuteSQL outputs Avro, so you're already halfway there.
2. Convert Avro to bulk UPSERT statements: Add a ConvertRecord processor with these configs:
   • Record Reader: Pick AvroReader to parse your input Avro records.
   • Record Writer: Select SQLRecordWriter, then tweak these key settings:
     • Set Statement Type to UPSERT (make sure your target DB supports UPSERT syntax—most modern RDBMS like PostgreSQL, MySQL, SQL Server do).
     • Specify your target Table Name (e.g., public.customer_data).
     • Enable Batch Mode to generate a single bulk UPSERT statement instead of individual ones (way faster for large datasets).
     • If your Avro field names don't match the target table columns, use the Column Names property to map them (e.g., avro_field_1:table_col1, avro_field_2:table_col2).
3. Run the UPSERTS: Pipe the output of ConvertRecord to a PutSQL processor configured with your local DB connection pool. PutSQL will execute the bulk statements directly.

No scripts, no manual parsing—all out-of-the-box NiFi magic.

思路2: Split records + parameterized UPSERT（无需ExecuteScript） #

If you need to go the single-record + parameterized query route, here's the simplest way to get past that Avro-to-sql.args conversion block:

1. Split Avro array into single records: Use SplitRecord with AvroReader as the Record Reader—it automatically splits the input Avro array into individual record FlowFiles.
2. Convert Avro to JSON: Add ConvertRecord again, this time using AvroReader and JsonRecordWriter. JSON is easier to extract field values/metadata from with NiFi's built-in tools.
3. Generate sql.args attributes: Use EvaluateJsonPath to pull field info and set the required parameters:
   • For each field (e.g., user_id of type INT), add these property mappings:
     • sql.args.1.name → user_id (matches the parameter name in your UPSERT statement)
     • sql.args.1.type → 4 (this is the JDBC type code for INT—look up your DB's JDBC type codes for other fields, e.g., VARCHAR is 12)
     • sql.args.1.value → $.user_id (JsonPath to pull the field value from the JSON content)
   • Repeat for all fields, incrementing the number in sql.args.N each time.
4. Set the UPSERT statement: Use UpdateAttribute to add a property sql.statement with your parameterized query, e.g.:

   UPSERT INTO your_local_table (user_id, username, email) VALUES (:user_id, :username, :email)
   

   The :param_name syntax matches the sql.args.N.name values you set earlier.
5. Execute the UPSERT: Send the FlowFiles to PutSQL—it'll automatically map the sql.args attributes to the statement parameters and run the query.

Pro tip for large field lists #

If you've got tons of fields and don't want to configure each one manually, use UpdateRecord with AvroReader and RecordPath expressions to dynamically generate sql.args attributes. For example, you can use a RecordPath like * to iterate all fields, then use NiFi's expression language to set:

• sql.args.${field.index}.name → ${field.name}
• sql.args.${field.index}.type → ${field.jdbcType} (this works if your Avro schema includes JDBC type metadata; if not, you can add a mapping lookup with LookupAttribute)

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