NiFi中CSV动态表头映射问题：实现向Druid的动态数据导入

Here's a practical, scalable workflow to handle variable CSV headers and import data into Druid:

1. Ingest CSV Files #

Use GetFile (or ListFile + FetchFile) to pull CSV files into NiFi. Add an UpdateAttribute processor to set attributes like csv.delimiter (e.g., ,) if your CSVs use a non-standard separator.

2. Split Header and Data Rows #

• First, use ExtractText to capture the header line as an attribute:
  • Set Regular Expression to ^(.*)$
  • Set Capture Group Attribute to csv.headers
  • Enable Line-by-Line Processing and set Maximum Lines to Process to 1
• Next, use SplitText to split the remaining content into individual data rows:
  • Set Line Split Count to 1
  • Enable Skip Header Line (since we already extracted it)
  • Add an attribute like row.content to store each row's value

3. Map CSV Headers to Druid Schema #

Create a flexible mapping between CSV headers and Druid's dimensions/metrics. For small, stable mappings, hardcode them in a script; for scalable updates, use a PropertyFileLookupService (store mappings in a properties file like user_id=user_id, transaction_time=__time).

Use ExecuteScript (Groovy) to generate a mapping attribute:

def headers = flowFile.getAttribute('csv.headers').split(flowFile.getAttribute('csv.delimiter'))
def mapping = [:]
// Define inline mappings or load from lookup service
def druidLookup = [
    "user_id": "user_id",
    "transaction_time": "__time",
    "total_revenue": "revenue",
    "product_name": "product_name"
]
headers.eachWithIndex { header, idx ->
    def cleanHeader = header.trim()
    if (druidLookup.containsKey(cleanHeader)) {
        mapping[idx] = druidLookup[cleanHeader]
    }
}
flowFile.setAttribute('header.mapping', new groovy.json.JsonBuilder(mapping).toString())
return flowFile

4. Convert Rows to Druid-Compatible JSON #

Use another ExecuteScript (Groovy) to turn each data row into a JSON object with mapped Druid fields:

def row = flowFile.getAttribute('row.content')
def delimiter = flowFile.getAttribute('csv.delimiter') ?: ','
def values = row.split(delimiter)
def mapping = new groovy.json.JsonSlurper().parseText(flowFile.getAttribute('header.mapping'))

def druidRecord = [:]
mapping.each { idx, druidField ->
    if (idx < values.size()) {
        def value = values[idx].trim()
        // Handle timestamp conversion for Druid's required __time field
        if (druidField == '__time') {
            // Adjust date format to match your CSV's timestamp
            def date = Date.parse("yyyy-MM-dd HH:mm:ss", value)
            druidRecord[druidField] = date.getTime() // Convert to epoch milliseconds
        } else {
            druidRecord[druidField] = value
        }
    }
}

// Write JSON to flow file content
flowFile = session.write(flowFile, { out ->
    out.write(new groovy.json.JsonBuilder(druidRecord).toString().getBytes('UTF-8'))
} as OutputStreamCallback)
flowFile.setAttribute('mime.type', 'application/json')
return flowFile

5. Batch Records for Druid Ingestion #

Druid performs best with batched data. Use MergeContent to group JSON records into a single file:

• Set Merge Strategy to Bin-Packing Algorithm
• Set Max Bin Size to a value like 1000000 (1MB)
• Set Delimiter Strategy to Text and Delimiter to \n (each JSON record on a new line)

6. Import into Druid #

Use PutDruidRecord to send the batched JSON to Druid:

• Configure your Druid coordinator/overlord endpoint
• Set Input Format to json
• Reference your Druid datasource and ingestion spec (ensure the spec accepts dynamic fields or uses the mapped dimensions/metrics)

Key Notes #

• Error Handling: Add RouteOnAttribute to route rows where values.size() != headers.size() to a failure queue for debugging.
• CSV Quoting: If your CSV has quoted values with commas, replace SplitText with ConvertRecord using a CSVReader that handles quoting.
• Scalability: For large mappings, replace inline mappings with a DistributedMapCache or PropertyFileLookupService to avoid script changes.

内容的提问来源于stack exchange，提问作者PM Ravi Teja