[Python] Apache Nifi: Unable to Merge Multiple CSVs into a Single PARQUET File using...

I am trying to merge the multiple CSVs which are coming from upstream as a flowfiles of similar kind (same Schema) into a one PARQUET file format. Below is the Flow of my processor group. Where in upper ExecuteStreamCommand i am renaming the column names to ensure there is no special character in the column names while in downstream ExecuteStreamCommand processor actually trying to merge in a single parquet format but it does not merge into single and the same number of CSV files (Which are in separate parquet files) comes out..



Here below is the code which i am using to merge multiple CSVs into single Parquet File.

import sys
import pandas as pd
import io
from pyarrow import parquet as pq
import pyarrow as pa

# Initialize an empty DataFrame to hold all CSV data
merged_df = pd.DataFrame()

# Read CSV data from standard input (incoming flow file content)
input_data = sys.stdin.read().strip()

# Check if the input data is empty
if not input_data:
print("Error: No data received from stdin")
sys.exit(1)

# Use StringIO to read the CSV from stdin
csv_content = io.StringIO(input_data)

# Read and append CSV content to merged_df
try:
# Read CSV into DataFrame
df = pd.read_csv(csv_content)

# If merged_df is empty, initialize it with the same columns as df
if merged_df.empty:
merged_df = df
else:
# Align columns before concatenating (this handles schema inconsistencies)
merged_df = pd.concat([merged_df, df], ignore_index=True, sort=False)

except pd.errors.EmptyDataError:
print("Error: No columns to parse from CSV data.")
sys.exit(1)

# After reading all CSV files, convert the merged DataFrame to a Parquet table
table = pa.Table.from_pandas(merged_df)

# Write the Parquet table to stdout (which NiFi will handle)
pq.write_table(table, sys.stdout.buffer, compression='snappy') # Adjust compression if needed


Can anybody please anticipate where i am doing it wrong? Why it is unable to merge into a single parquet rather multiple Parquet file. It is also not changing the extension of output files like .parquet.

Continue reading...