连接器

译者：flink.sojb.cn

从文件系统中读取

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

使用Hadoop文件系统实现

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

• S3（已测试）
• 适用于Hadoop的Google云端存储连接器（已测试）
• Alluxio（已测试）
• XtreemFS（已测试）
• FTP通过Hftp（未测试）
• 还有很多。

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

• 在flink-conf.yaml已设定的fs.hdfs.hadoopconf属性将Hadoop配置目录。对于自动测试或从IDE运行，flink-conf.yaml可以通过定义FLINK_CONF_DIR环境变量来设置包含的目录。
• Hadoop配置（在该目录中）具有文件中所需文件系统的条目core-site.xml。S3和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兼容性的更多信息。

Avro对Flink的支持

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开始）。