Parquet format

Flink supports reading Parquet files and producing Flink rows. To use the format you need to add the Flink Parquet dependency to your project:

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-parquet</artifactId>
  4.     <version>1.14.4</version>
  5. </dependency>

This format is compatible with the new Source that can be used in both batch and streaming modes. Thus, you can use this format in two ways:

  • Bounded read for batch mode
  • Continuous read for streaming mode: monitors a directory for new files that appear

Bounded read example:

In this example we create a DataStream containing Parquet records as Flink Rows. We project the schema to read only certain fields (“f7”, “f4” and “f99”).
We read records in batches of 500 records. The first boolean parameter specifies if timestamp columns need to be interpreted as UTC. The second boolean instructs the application if the projected Parquet fields names are to be interpreted in a case sensitive way. There is no need for a watermark strategy as records do not contain event timestamps.

  1. final LogicalType[] fieldTypes =
  2.   new LogicalType[] {
  3.   new DoubleType(), new IntType(), new VarCharType()
  4.   };
  6. final ParquetColumnarRowInputFormat<FileSourceSplit> format =
  7.   new ParquetColumnarRowInputFormat<>(
  8.   new Configuration(),
  9.   RowType.of(fieldTypes, new String[] {"f7", "f4", "f99"}),
  10.   500,
  11.   false,
  12.   true);
  13. final FileSource<RowData> source =
  14.   FileSource.forBulkFileFormat(format,  /* Flink Path */)
  15.   .build();
  16. final DataStream<RowData> stream =
  17.   env.fromSource(source, WatermarkStrategy.noWatermarks(), "file-source");

Continuous read example:

In this example we create a DataStream containing Parquet records as Flink Rows that will infinitely grow as new files are added to the directory. We monitor for new files each second. We project the schema to read only certain fields (“f7”, “f4” and “f99”).
We read records in batches of 500 records. The first boolean parameter specifies if timestamp columns need to be interpreted as UTC. The second boolean instructs the application if the projected Parquet fields names are to be interpreted in a case sensitive way. There is no need for a watermark strategy as records do not contain event timestamps.

  1. final LogicalType[] fieldTypes =
  2.   new LogicalType[] {
  3.   new DoubleType(), new IntType(), new VarCharType()
  4.   };
  6. final ParquetColumnarRowInputFormat<FileSourceSplit> format =
  7.   new ParquetColumnarRowInputFormat<>(
  8.   new Configuration(),
  9.   RowType.of(fieldTypes, new String[] {"f7", "f4", "f99"}),
  10.   500,
  11.   false,
  12.   true);
  13. final FileSource<RowData> source =
  14.   FileSource.forBulkFileFormat(format,  /* Flink Path */)
  15.   .monitorContinuously(Duration.ofSeconds(1L))
  16.   .build();
  17. final DataStream<RowData> stream =
  18.   env.fromSource(source, WatermarkStrategy.noWatermarks(), "file-source");