[Python] Pyflink app outputs [B@... instead of expected csv rows

python version: 3.11.9
apache-flink version: 2.0

I am trying to learn pyflink by working on a small application that

1. Has an input source WeatherMeasurements.csv with the following columns ['city', 'state', 'county', 'avgTemp', 'avgHumidity', 'avgPressure', 'sampleTime']
2. Parses each line, convert avgTemp to celsius and compute comfort index, append both results to line.
3. Write results to output.csv

Here is an example of what WeatherMeasurements.csv looks like

Madison,Wisconsin,Dane,72.2,39.94,950.89,2025-07-02T13:13:12.051886
Houston,Texas,Harris,47.03,60.53,1005.75,2025-07-02T13:14:12.051886
Madison,Wisconsin,Dane,47.26,46.11,971.65,2025-07-02T13:15:12.051886
...


After the transformations are applied I expecting the output.csv file to look something like this:

Expected output.csv

Madison,Wisconsin,Dane,72.20,39.94,950.89,2025-07-02T13:13:12.051886,22.33,43.36
Houston,Texas,Harris,47.03,60.53,1005.75,2025-07-02T13:14:12.051886,8.35,18.56
Madison,Wisconsin,Dane,47.26,46.11,971.65,2025-07-02T13:15:12.051886,8.48,25.47
...


but instead, it looks like its just printing references to the data

Actual output.csv

[B@dba59aa
[B@3be0300f
[B@7b1be48d
...


Here is my code, I would like some assistance as to why I am not getting the expected output.

from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.common import Encoder, WatermarkStrategy
from pyflink.datastream.connectors.file_system import FileSink, FileSource, StreamFormat, RollingPolicy, OutputFileConfig

# Construct a dictionary for each line, used to make downstream operations simpler
def parse(line):
city, state, county, t, h, p, ts = line.split(",")
return {
"city": city,
"state": state,
"county": county,
"avgTemp": float(t),
"avgHumidity": float(h),
"avgPressure": float(p),
"sampleTime": ts
}

# Compute comfort index and append to row
def enrich(rec):
rec["avgTempC"] = (rec["avgTemp"] - 32) * 5/9
rec["comfortIndex"] = rec["avgTemp"] * (1 - rec["avgHumidity"]/100)
return rec

# after enrich, map to a CSV-formatted string
def format_csv(rec):
csv_line = ",".join([
rec["city"],
rec["state"],
rec["county"],
f"{rec['avgTemp']:.2f}",
f"{rec['avgHumidity']:.2f}",
f"{rec['avgPressure']:.2f}",
rec["sampleTime"],
f"{rec['avgTempC']:.2f}",
f"{rec['comfortIndex']:.2f}"
])
return csv_line + "\n"

def main(file_path):
env = StreamExecutionEnvironment.get_execution_environment()
env.set_runtime_mode(RuntimeExecutionMode.BATCH)
env.set_parallelism(1)

file_src = (
FileSource
.for_record_stream_format(StreamFormat.text_line_format(), file_path)
.process_static_file_set()
.build()
)

lines = env.from_source(
source=file_src,
watermark_strategy=WatermarkStrategy.no_watermarks(),
source_name="weather-file-source"
)

parsed = lines.map(parse)
enriched = parsed.map(enrich)
csv_lines = enriched.map(format_csv)

csv_lines.sink_to(
sink=FileSink.for_row_format(
base_path="output",
encoder=Encoder.simple_string_encoder())
.with_output_file_config(
OutputFileConfig.builder()
.with_part_prefix("weather_")
.with_part_suffix(".csv")
.build())
.with_rolling_policy(RollingPolicy.default_rolling_policy())
.build()
)

print("Writing to output directory")
env.execute("Weather Enrichment Job")

if __name__ == '__main__':
file_path = "WeatherMeasurements.csv"
main(file_path)

Continue reading...