コネクタ

ファイルシステムからの読み込み

Flinkは以下のファイルシステムのための組み込みのサポートを持ちます:

ファイルシステム スキーマ 備考
Hadoop 分散型ファイルシステム (HDFS)   hdfs:// 全ての HDFS バージョンがサポートされます
Amazon S3 s3:// Hadoopファイルシステム実装を使ったサポート (以下を見てください)
MapR ファイルシステム maprfs:// ユーザはlib/ ディレクトリ内に手動で必要なjarファイルを置く必要があります
Alluxio alluxio://   Hadoopファイルシステム実装を使ったサポート (以下を見てください)

Hadoopファイルシステム実装の使用

Apache Flink を使ってユーザはorg.apache.hadoop.fs.FileSystem インタフェースを実装する任意のファイルシステムを使うことができます。Hadoop FileSystem 実装については

FlinkでHadoopファイルシステムを使うには、以下を確認してください。

  • flink-conf.yaml でHadoop設定ディレクトリへのfs.hdfs.hadoopconf プロパティが設定されている。自動化テストあるいはIDEから実行するために、flink-conf.yaml を含むディレクトリはFLINK_CONF_DIR 環境変数によって定義することができます。
  • (そのディレクトリ内の)Hadoop 設定はファイルcore-site.xml内に必要とされるファイルシステムのためのエントリを持ちます。S3 と Alluxio の例は以下でリンク/示されます。
  • ファイルシステムを使うために必要とされるクラスは(Flinkが動いている全てのマシーン上の)Flinkのインストレーションのlib/ フォルダで利用可能です。ファイルをディレクトリに置くことは不可能であれば、FlinkはHadoop jarファイルにクラスパスを追加するためにHADOOP_CLASSPATH 環境変数を期待します。

Amazon S3

利用可能なS3ファイルシステム実装、それらの設定および必要なライブラリについては、配備 & オペレーション - 配備 - AWS - S3: Simple Storage Service を見てください。

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はInputFormatOutputFormatと呼ばれる概念を持ちます。

InputFormatの1つの実装にHadoopInputFormatがあります。これはユーザが全ての既存のHadoop入力フォーマットをFlinkで使うことができるラッパーです。

この章ではFlinkを他のシステムに接続するいくつかの例を示します。FlinkのHadoop互換性についてはもっと読む

FlinkはApache Avroのための広範囲に及ぶ組み込みサポートを持ちます。これによりFlinkを使ってAvroファイルから簡単に読むことができます。また、Flinkのシリアライズ化フレームワークはAvroスキーマから生成されたクラスを扱うことができます。

Avroファイルからデータを読み込むためには、AvroInputFormatを指定する必要があります。

:

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

User はAvroで生成された POJO であることに注意してください。FlinkはこれらのPOJOの文字ベースキーの選択をすることもできます。例えば:

usersDS.groupBy("name")

FlinkでのGenericData.Record 型の利用は可能ですが、お勧めでは無いことに注意してください。Since the record contains the full schema, its very data intensive and thus probably slow to use.

Flinkの POJO フィールドの選択もAvroから生成されたPOJOと連携します。しかし、それはフィールドの型が生成されたクラスへ正しく書かれている場合のみ可能です。フィールドがObject型の場合、フィールドをjoinあるいはgroupingキーとして使うことができません。Specifying a field in Avro like this {"name": "type_double_test", "type": "double"}, works fine, however specifying it as a UNION-type with only one field ({"name": "type_double_test", "type": ["double"]},) will generate a field of type Object. nullが可能な型 ({"name": "type_double_test", "type": ["null", "double"]},) の指定が可能なことに注意してください!

Microsoft Azure Table Storageへのアクセス

注意: この例はFlink 0.6-incubating から動作します

この例では Azureのテーブルストレージにアクセスするための既存のHadoop入力フォーマットの実装を使うために HadoopInputFormat を使っています。

  1. azure-tables-hadoop プロジェクトをダウンロードしコンパイルします。プロジェクトによって開発された入力フォーマットはまだMaven Centralで利用可能ではありません。従って私たち自身でプロジェクトをビルドする必要があります。以下のコマンドを実行します:

    git clone https://github.com/mooso/azure-tables-hadoop.git
    cd azure-tables-hadoop
    mvn clean install
  2. クイックスタートを使って新しいFlinkのプロジェクトをセットアップする:

    curl https://flink.apache.org/q/quickstart.sh | bash
  3. pom.xmlファイルに以下の依存を追加します (<dependencies> セクションの中):

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-hadoop-compatibility_2.10</artifactId>
        <version>1.3-SNAPSHOT</version>
    </dependency>
    <dependency>
      <groupId>com.microsoft.hadoop</groupId>
      <artifactId>microsoft-hadoop-azure</artifactId>
      <version>0.0.4</version>
    </dependency>

    flink-hadoop-compatibility is a Flink package that provides the Hadoop input format wrappers. microsoft-hadoop-azure は私たちが私たちのプロジェクトのためにビルドしたプロジェクトを追加しています。

これでプロジェクトはコードを開始する準備ができました。プロジェクトを、EclipseあるいはIntelliJのようなIDEにインポートすることをお勧めします。(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を使って、全ての既知の変換をデータセットに適用することができます。

MongoDBのアクセス

This GitHub repository documents how to use MongoDB with Apache Flink (starting from 0.7-incubating).

TOP
inserted by FC2 system