Hadoop formats

Project Configuration

对 Hadoop 的支持位于 flink-hadoop-compatibility Maven 模块中。

将以下依赖添加到 pom.xml 中使用 hadoop

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-hadoop-compatibility_2.12</artifactId>
  4.     <version>1.18.1</version>
  5. </dependency>

如果你想在本地运行你的 Flink 应用（例如在 IDE 中），你需要按照如下所示将 hadoop-client 依赖也添加到 pom.xml：

  1. <dependency>
  2.     <groupId>org.apache.hadoop</groupId>
  3.     <artifactId>hadoop-client</artifactId>
  4.     <version>2.10.2</version>
  5.     <scope>provided</scope>
  6. </dependency>

Using Hadoop InputFormats

在 Flink 中使用 Hadoop InputFormats，必须首先使用 HadoopInputs 工具类的 readHadoopFile 或 createHadoopInput 包装 Input Format。 前者用于从 FileInputFormat 派生的 Input Format，而后者必须用于通用的 Input Format。 生成的 InputFormat 可通过使用 ExecutionEnvironment#createInput 创建数据源。

生成的 DataStream 包含 2 元组，其中第一个字段是键，第二个字段是从 Hadoop InputFormat 接收的值。

下面的示例展示了如何使用 Hadoop 的 TextInputFormat。

Java

  1. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  2.   KeyValueTextInputFormat textInputFormat = new KeyValueTextInputFormat();
  4.   DataStream<Tuple2<Text, Text>> input = env.createInput(HadoopInputs.readHadoopFile(
  5.   textInputFormat, Text.class, Text.class, textPath));
  7.   // Do something with the data.
  8.   [...]

Scala

  1. val env = StreamExecutionEnvironment.getExecutionEnvironment
  2. val textInputFormat = new KeyValueTextInputFormat
  3. val input: DataStream[(Text, Text)] =
  4.   env.createInput(HadoopInputs.readHadoopFile(
  5.     textInputFormat, classOf[Text], classOf[Text], textPath))
  7.     // Do something with the data.
  8.     [...]