Flinkは以下のファイルシステムのための組み込みのサポートを持ちます:
ファイルシステム | スキーマ | 備考 |
---|---|---|
Hadoop 分散型ファイルシステム (HDFS) | hdfs:// |
全ての HDFS バージョンがサポートされます |
Amazon S3 | s3:// |
Hadoopファイルシステム実装を使ったサポート (以下を見てください) |
MapR ファイルシステム | maprfs:// |
ユーザはlib/ ディレクトリ内に手動で必要なjarファイルを置く必要があります |
Alluxio | alluxio:// |
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
環境変数によって定義することができます。core-site.xml
内に必要とされるファイルシステムのためのエントリを持ちます。S3 と Alluxio の例は以下でリンク/示されます。lib/
フォルダで利用可能です。ファイルをディレクトリに置くことは不可能であれば、FlinkはHadoop jarファイルにクラスパスを追加するためにHADOOP_CLASSPATH
環境変数を期待します。利用可能なS3ファイルシステム実装、それらの設定および必要なライブラリについては、配備 & オペレーション - 配備 - AWS - S3: Simple Storage Service を見てください。
Alluxio サポートのために、以下のエントリをcore-site.xml
ファイルに追加してください:
<property>
<name>fs.alluxio.impl</name>
<value>alluxio.hadoop.FileSystem</value>
</property>
Apache Flinkによってユーザは多くの異なるシステムにデータソースあるいはシンクとしてアクセスすることができます。システムはとても簡単に拡張できるように設計されます。Apache Hadoopと似て、FlinkはInputFormat
とOutputFormat
と呼ばれる概念を持ちます。
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"]},
) の指定が可能なことに注意してください!
注意: この例はFlink 0.6-incubating から動作します
この例では Azureのテーブルストレージにアクセスするための既存のHadoop入力フォーマットの実装を使うために HadoopInputFormat
を使っています。
azure-tables-hadoop
プロジェクトをダウンロードしコンパイルします。プロジェクトによって開発された入力フォーマットはまだMaven Centralで利用可能ではありません。従って私たち自身でプロジェクトをビルドする必要があります。以下のコマンドを実行します:
git clone https://github.com/mooso/azure-tables-hadoop.git
cd azure-tables-hadoop
mvn clean install
クイックスタートを使って新しいFlinkのプロジェクトをセットアップする:
curl https://flink.apache.org/q/quickstart.sh | bash
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
はHadoop入力フォーマットのラッパーを提供するFlinkパッケージです。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
を使って、全ての既知の変換をデータセットに適用することができます。
このGitHub リポジトリはMongoDBをApache Flink(0.7-incubatingから)と一緒に使う方法を説明します。