Использовать файлы GCS на основе шаблона из Flink - PullRequest
Новая
       26

Использовать файлы GCS на основе шаблона из Flink

1 голос
/ 10 апреля 2019

Поскольку Flink поддерживает абстракцию файловой системы Hadoop и имеется GCS-соединитель - библиотека, реализующая его поверх Google Cloud Storage.

Как мне создать исходный файл Flink с помощью кода в этом репо?

1 Ответ

2 голосов
/ 10 апреля 2019

Для этого нужно:

  1. Установите и настройте разъем GCS на вашем кластере Flink.
  2. Добавьте зависимости Hadoop и Flink (включая HDFS-разъем ) в ваш проект: 
    <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-scala_2.11</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-filesystem_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-hadoop-compatibility_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>${hadoop.version}</version>
    <scope>provided</scope>
</dependency>
<dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-core</artifactId>
    <version>${hadoop.version}</version>
    <scope>provided</scope>
</dependency>

  3. Используйте его для создания источника данных с путем GCS: 

    import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.hadoopcompatibility.HadoopInputs;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.TextInputFormat;

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

DataSet<Tuple2<LongWritable, Text>> input =
    env.createInput(
        HadoopInputs.readHadoopFile(
            new TextInputFormat(), LongWritable.class, Text.class, "gs://bucket/path/some*pattern/"));
...