Azure Table Storage

本例使用 HadoopInputFormat 包装器来使用现有的 Hadoop input format 实现访问 Azure’s Table Storage.

  1. 下载并编译 azure-tables-hadoop 项目。该项目开发的 input format 在 Maven 中心尚不存在，因此，我们必须自己构建该项目。 执行如下命令：
  1. git clone https://github.com/mooso/azure-tables-hadoop.git
  2. cd azure-tables-hadoop
  3. mvn clean install
  1. 使用 quickstarts 创建一个新的 Flink 项目：
  1. curl https://flink.apache.org/q/quickstart.sh | bash
  1. 在你的 pom.xml 文件 <dependencies> 部分添加如下依赖：
  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-hadoop-compatibility_2.12</artifactId>
  4.     <version>1.16.0</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>com.microsoft.hadoop</groupId>
  8.     <artifactId>microsoft-hadoop-azure</artifactId>
  9.     <version>0.0.5</version>
  10. </dependency>

flink-hadoop-compatibility 是一个提供 Hadoop input format 包装器的 Flink 包。 microsoft-hadoop-azure 可以将之前构建的部分添加到项目中。

现在可以开始进行项目的编码。我们建议将项目导入 IDE，例如 IntelliJ。你应该将其作为 Maven 项目导入。 跳转到文件 Job.java。这是 Flink 作业的初始框架。

粘贴如下代码：

  1. import java.util.Map;
  2. import org.apache.flink.api.common.functions.MapFunction;
  3. import org.apache.flink.api.java.DataStream;
  4. import org.apache.flink.api.java.ExecutionEnvironment;
  5. import org.apache.flink.api.java.tuple.Tuple2;
  6. import org.apache.flink.hadoopcompatibility.mapreduce.HadoopInputFormat;
  7. import org.apache.hadoop.io.Text;
  8. import org.apache.hadoop.mapreduce.Job;
  9. import com.microsoft.hadoop.azure.AzureTableConfiguration;
  10. import com.microsoft.hadoop.azure.AzureTableInputFormat;
  11. import com.microsoft.hadoop.azure.WritableEntity;
  12. import com.microsoft.windowsazure.storage.table.EntityProperty;
  14. public class AzureTableExample {
  16.   public static void main(String[] args) throws Exception {
  17.     // 安装 execution environment
  18.     final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  20.     env.setRuntimeMode(RuntimeExecutionMode.BATCH);
  21.     // 使用 Hadoop input format 包装器创建 AzureTableInputFormat
  22.     HadoopInputFormat<Text, WritableEntity> hdIf = new HadoopInputFormat<Text, WritableEntity>(new AzureTableInputFormat(), Text.class, WritableEntity.class, new Job());
  24.     // 设置 Account URI，如 https://apacheflink.table.core.windows.net
  25.     hdIf.getConfiguration().set(azuretableconfiguration.Keys.ACCOUNT_URI.getKey(), "TODO");
  26.     // 设置存储密钥
  27.     hdIf.getConfiguration().set(AzureTableConfiguration.Keys.STORAGE_KEY.getKey(), "TODO");
  28.     // 在此处设置表名
  29.     hdIf.getConfiguration().set(AzureTableConfiguration.Keys.TABLE_NAME.getKey(), "TODO");
  31.     DataStream<Tuple2<Text, WritableEntity>> input = env.createInput(hdIf);
  32.     // 如何在 map 中使用数据的简单示例。
  33.     DataStream<String> fin = input.map(new MapFunction<Tuple2<Text,WritableEntity>, String>() {
  34.       @Override
  35.       public String map(Tuple2<Text, WritableEntity> arg0) throws Exception {
  36.         System.err.println("--------------------------------\nKey = "+arg0.f0);
  37.         WritableEntity we = arg0.f1;
  39.         for(Map.Entry<String, EntityProperty> prop : we.getProperties().entrySet()) {
  40.           System.err.println("key="+prop.getKey() + " ; value (asString)="+prop.getValue().getValueAsString());
  41.         }
  43.         return arg0.f0.toString();
  44.       }
  45.     });
  47.     // 发送结果（这仅在本地模式有效）
  48.     fin.print();
  50.     // 执行程序
  51.     env.execute("Azure Example");
  52.   }
  53. }

该示例展示了如何访问 Azure 表和如何将数据转换为 Flink 的 DataStream（更具体地说，集合的类型是 DataStream<Tuple2<Text, WritableEntity>>）。你可以将所有已知的 transformations 应用到 DataStream 实例。