Hey everyone,
TLDR: How to skip header or get head...
# randomb
Bhupendra Yadav
08/02/2023, 8:54 AMHey everyone,
TLDR: How to skip header or get headers when using CsvReaderFormat?
When using CsvReaderFormat, is there any way we can skip reading the header? I was looking at the source code to see If I can extend CsvReaderFormat but could see the file reading is not happening in it and hence couldn't figure out a way.
Also, I see that functionality to skip header in CsvReader but that requires ExecutionEnvironment while we are trying to use StreamExecutionEnvironment.
Any suggestions are appreciated.
Bhupendra Yadav
08/02/2023, 1:31 PMI found a way. We can use flink forSchema method and pass our own mapper
Copy code
import java.io.File;
import java.util.Map;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.csv.CsvReaderFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.jackson.JacksonMapperFactory;
public class CsvTest {
public static void main(String[] args) throws Exception {
String inputFile = "csv-path";
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Map<String, String> map = Map.ofEntries(Map.entry("jobName", "test-csv"));
ParameterTool parameterTool2 = ParameterTool.fromMap(map);
env.getConfig().setGlobalJobParameters(parameterTool2);
CsvReaderFormat<User> csvReaderFormat =
CsvReaderFormat.forSchema(
() -> JacksonMapperFactory.createCsvMapper(),
mapper -> mapper.schemaFor(User.class).withHeader().withoutQuoteChar(), // you can pass here with/without header
TypeInformation.of(User.class));
FileSource fs =
FileSource.forRecordStreamFormat(csvReaderFormat, Path.fromLocalFile(new File(inputFile)))
.build();
env.fromSource(fs, WatermarkStrategy.noWatermarks(), "CSV Source").print();
env.execute();
}
}