This documentation is for an unreleased version of Apache Flink. We recommend you use the latest stable version.
Azure Table storage
Azure Table Storage #
この例では、HadoopInputFormat
ラッパーを使ってAzureのTable Storageにアクセスするための既存のHadoop入力形式の実装を使います。
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.12</artifactId>
<version>1.19-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.microsoft.hadoop</groupId>
<artifactId>microsoft-hadoop-azure</artifactId>
<version>0.0.5</version>
</dependency>
flink-hadoop-compatibility
はHadoop入力フォーマットのラッパーを提供するFlinkパッケージです。
microsoft-hadoop-azure
は、以前にビルドしたプロジェクトを、プロジェクトに追加しています。
これで、プロジェクトのコーディングを開始する準備が整いました。プロジェクトをIntelliJのようなIDEにインポートすることをお勧めします。これをMavenプロジェクトとしてインポートする必要があります。
ファイルJob.java
を参照します。これはFlinkジョブの空のスケルトンです。
次のコードを貼り付けます:
import java.util.Map;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataStream;
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 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
// 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");
DataStream<Tuple2<Text, WritableEntity>> input = env.createInput(hdIf);
// a little example how to use the data in a mapper.
DataStream<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のDataStream
(より具合的には、セットの型はDataStream<Tuple2<Text, WritableEntity>>
)に変換する方法を説明します。DataStream
を使うと、既知の変換を全てDataStreamに適用できます。