Azure Table storage
This documentation is for an unreleased version of Apache Flink. We recommend you use the latest stable version.

Azure Table Storage #

この例では、HadoopInputFormatラッパーを使ってAzureのTable Storageにアクセスするための既存のHadoop入力形式の実装を使います。

  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. 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に適用できます。

Back to top

inserted by FC2 system