连接器

从文件系统中读取

Flink内置支持以下文件系统:

文件系统 方案 笔记
Hadoop分布式文件系统(HDFS)   hdfs:// 支持所有HDFS版本
亚马逊S3 s3:// 通过Hadoop文件系统实现支持(见下文)
MapR文件系统 maprfs:// 用户必须手动将所需的jar文件放在lib/dir中
Alluxio alluxio://   通过Hadoop文件系统实现支持(见下文)

使用Hadoop文件系统实现

Apache Flink允许用户使用任何实现该org.apache.hadoop.fs.FileSystem 接口的文件系统有Hadoop FileSystem实现

为了使用Flink的Hadoop文件系统,请确保

  • flink-conf.yaml已设定的fs.hdfs.hadoopconf属性将Hadoop配置目录。对于自动测试或从IDE运行,flink-conf.yaml可以通过定义FLINK_CONF_DIR环境变量来设置包含的目录
  • Hadoop配置(在该目录中)具有文件中所需文件系统的条目core-site.xmlS3和Alluxio的示例链接/显示如下。
  • lib/Flink安装文件夹中提供了使用文件系统所需的类(在运行Flink的所有计算机上)。如果无法将文件放入目录,Flink还会尊重HADOOP_CLASSPATH环境变量以将Hadoop jar文件添加到类路径中。

亚马逊S3

请参阅部署和 算子操作 - 部署 - AWS - S3:简单存储服务,以获取可用的S3文件系统实现,其配置和所需的库。

Alluxio

对于Alluxio支持,将以下条目添加到core-site.xml文件中:

<property>
  <name>fs.alluxio.impl</name>
  <value>alluxio.hadoop.FileSystem</value>
</property>

使用Hadoop的Input / OutputFormat打包器连接到其他系统

Apache Flink允许用户访问许多不同的系统作为数据源或接收器。该系统的设计非常容易扩展。与Apache Hadoop类似,Flink具有所谓的InputFormats和OutputFormats 的概念

这些InputFormat一个实现HadoopInputFormat这是一个打包器,允许用户使用Flink的所有现有Hadoop输入格式。

本节介绍将Flink连接到其他系统的一些示例。 阅读有关Flink中Hadoop兼容性的更多信息

Flink对Apache Avro提供了广泛的内置支持这样可以使用Flink轻松读取Avro文件。此外,Flink的序列化框架能够处理从Avro架构生成的类。确保将Flink Avro依赖项包含在项目的pom.xml中。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-avro</artifactId>
  <version>1.7-SNAPSHOT</version>
</dependency>

要从Avro文件中读取数据,您必须指定一个AvroInputFormat

示例

AvroInputFormat<User> users = new AvroInputFormat<User>(in, User.class);
DataSet<User> usersDS = env.createInput(users);

请注意,这User是Avro生成的POJO。Flink还允许执行这些POJO的基于字符串的Keys选择。例如:

usersDS.groupBy("name")

请注意,使用GenericData.RecordFlink可以使用该类型,但不建议使用。由于记录包含完整的模式,因此其数据密集,因此可能使用起来很慢。

Flink的POJO字段选择也适用于Avro生成的POJO。但是,只有在将字段类型正确写入生成的类时才可以使用。如果字段是类型Object,则不能将该字段用作连接或分组键。像这样在Avro中指定一个字段{"name": "type_double_test", "type": "double"},工作正常,但是将其指定为只有一个字段({"name": "type_double_test", "type": ["double"]},的UNION类型将生成一个类型的字段Object请注意,指定可空类型({"name": "type_double_test", "type": ["null", "double"]},)是可能的!

访问Microsoft Azure表存储

注意:此示例适用于Flink 0.6-incubating

此示例使用HadoopInputFormat打包器使用现有的Hadoop输入格式实现来访问Azure的表存储

  1. 下载并编译azure-tables-hadoop项目。该项目开发的输入格式尚未在Maven Central中提供,因此,我们必须自己构建项目。执行以下命令:
   git clone https://github.com/mooso/azure-tables-hadoop.git
   cd azure-tables-hadoop
   mvn clean install
   
  1. 使用快速入门设置新的Flink项目:
   curl https://flink.apache.org/q/quickstart.sh | bash
   
  1. 将以下依赖项(在本<dependencies>节中)添加到您的pom.xml文件中:
   <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-hadoop-compatibility_2.11</artifactId>
       <version>1.7-SNAPSHOT</version>
   </dependency>
   <dependency>
     <groupId>com.microsoft.hadoop</groupId>
     <artifactId>microsoft-hadoop-azure</artifactId>
     <version>0.0.4</version>
   </dependency>
   

flink-hadoop-compatibility是一个Flink包,提供Hadoop输入格式打包器。 microsoft-hadoop-azure将我们之前构建的项目添加到项目中。

该项目现在准备开始编码。我们建议将项目导入IDE,例如Eclipse或IntelliJ。(作为Maven项目导入!)。浏览到该Job.java文件的代码它是Flink工作的空框架。

将以下代码粘贴到其中:

import java.util.Map;
import org.apache.flink.api.common.functions.MapFunction;
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.mapreduce.HadoopInputFormat;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import com.microsoft.hadoop.azure.AzureTableConfiguration;
import com.microsoft.hadoop.azure.AzureTableInputFormat;
import com.microsoft.hadoop.azure.WritableEntity;
import com.microsoft.windowsazure.storage.table.EntityProperty;

public class AzureTableExample {

  public static void main(String[] args) throws Exception {
    // set up the execution environment
    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    // create a  AzureTableInputFormat, using a Hadoop input format wrapper
    HadoopInputFormat<Text, WritableEntity> hdIf = new HadoopInputFormat<Text, WritableEntity>(new AzureTableInputFormat(), Text.class, WritableEntity.class, new Job());

    // set the Account URI, something like: https://apacheflink.table.core.windows.net
    hdIf.getConfiguration().set(AzureTableConfiguration.Keys.ACCOUNT_URI.getKey(), "TODO");
    // set the secret storage key here
    hdIf.getConfiguration().set(AzureTableConfiguration.Keys.STORAGE_KEY.getKey(), "TODO");
    // set the table name here
    hdIf.getConfiguration().set(AzureTableConfiguration.Keys.TABLE_NAME.getKey(), "TODO");

    DataSet<Tuple2<Text, WritableEntity>> input = env.createInput(hdIf);
    // a little example how to use the data in a mapper.
    DataSet<String> fin = input.map(new MapFunction<Tuple2<Text,WritableEntity>, String>() {
      @Override
      public String map(Tuple2<Text, WritableEntity> arg0) throws Exception {
        System.err.println("--------------------------------\nKey = "+arg0.f0);
        WritableEntity we = arg0.f1;

        for(Map.Entry<String, EntityProperty> prop : we.getProperties().entrySet()) {
          System.err.println("key="+prop.getKey() + " ; value (asString)="+prop.getValue().getValueAsString());
        }

        return arg0.f0.toString();
      }
    });

    // emit result (this works only locally)
    fin.print();

    // execute program
    env.execute("Azure Example");
  }
}

该示例显示了如何访问Azure表并将数据转换为Flink DataSet(更具体地说,是集合的类型DataSet<Tuple2<Text, WritableEntity>>)。使用DataSet,您可以将所有已知的转换应用于DataSet。

访问MongoDB

这个GitHub存储库记录了如何将MongoDB与Apache Flink一起使用(从0.7-incubating开始)

回到顶部