Hi Everyone, Is there functionality in Flink like caching an object (e.g. from a process function) and accessing it elsewhere? Our use-case: we are reading a CSV file with headers since flink does not support writing back headers to output files(at least we couldn't figure out this), we are thinking to solve it like this: 1. Read the CSV file as Map(key as header and value as actual column value) and add the headers(once) in the cache while processing. 2. Before writing the actual output file, read the headers from the cache and add them to the file. We are doing this using TextOutputFormat.open with parallelism always 1. Any suggestions or easier ways to do this will be appreciated. We are aware of Flink Distributed cache where we can register a file as cache and read it later, but this file is only accessible in RichMapFunctions and in our case while writing we don't have access to runtimeContext.

Can you explain on a more high level what you are trying to achieve? Do you have a batch processing use case, or streaming? Does the CSV file contain metadata necessary for processing large amounts of data, or is the CSV file the data you want to process with Flink?

Our usecase is simple. We have a CSV file with headers, we want to read the CSV file -> process/enrich records based on business logic and write the data to an output file with headers.

What's the size of the CSV file? I wonder if it isn't easier to process this in plain Java, as you don't really need all the parallelization features of Flink. If the CSV file is say just 100gb or so, then a standalone Java app is probably simple to implement and quite efficient. If you are going with Flink: For these kinds of problems, there are 2 strategies: Either you ship the schema (CSV file headers) with each record through the system -- this is inefficient but easy to implement (e.g. each event is a map with header / value) Or you read and ship the schema once and each record flows w/o the schema through the system -- efficient but more involved to implement. (in the Flink main method you read the schema, ship it to all the operators and then send the data through) The CSV reader (and writer) in Flink assume that you put the schema into the Java code. If you want to make this dynamic, e.g. read the schema from the file every time, you'll need to implement custom sources and sinks.

Thanks for your response, it makes sense. Our files aren't very large. Max a few gb's. So We were able to make it work by passing pojo class type as Java map. Then it passes headers in each record. Before writing we run a flaatmap, that extracts headers from a record(one time) and add it to stream and write the stream to a file

Okay nice -- I'm happy to hear that you've found a solution that works for you