Tablestore:チュートリアル

データの準備

Tablestoreにpetという名前のデータテーブルを準備します。name列は唯一のプライマリキー列です。次の表に、テーブル内のデータを示します。

Hiveを使用してTablestoreテーブルにアクセスする

1. 次のコマンドを実行して、/etc/profileにHADOOP_HOMEとHADOOP_CLASSPATHを追加します。

   export HADOOP_HOME=${Your Hadoop installation directory}
   export HADOOP_CLASSPATH=emr-tablestore-1.4.2.jar:tablestore-4.3.1-jar-with-dependencies.jar:joda-time-2.9.4.jar

2. bin/hiveコマンドを実行してHiveに移動し、外部テーブルを作成します。SQLステートメントの例：

   CREATE EXTERNAL TABLE pet
     (name STRING, owner STRING, species STRING, sex STRING, birth STRING, death STRING)
     STORED BY 'com.aliyun.openservices.tablestore.hive.TableStoreStorageHandler'
     WITH SERDEPROPERTIES(
       "tablestore.columns.mapping"="name,owner,species,sex,birth,death")
     TBLPROPERTIES (
       "tablestore.endpoint"="YourEndpoint",
       "tablestore.access_key_id"="YourAccessKeyId",
       "tablestore.access_key_secret"="YourAccessKeySecret",
       "tablestore.table.name"="pet");

   次の表に、外部テーブルの作成時に設定できるパラメーターを示します。

   パラメーター	説明
   WITH SERDEPROPERTIES	フィールドマッピングの設定。tablestore.columns.mappingの設定が含まれます。 デフォルトでは、外部テーブルのフィールド名は、Tablestoreテーブルのプライマリキー列または属性列の名前と同じです。外部テーブルのフィールド名がTablestoreテーブルの列名と異なる場合は、tablestore.columns.mappingを指定する必要があります。このパラメーターの値は、カンマ区切りの文字列です。カンマ(,)の両端にスペースを入れることはできません。各項目は、Tablestoreテーブルの列名を指定します。列名の順序は、外部テーブルのフィールド名の順序と同じです。 説明 Tablestoreの列名には空白文字を含めることができます。そのため、空白文字はTablestoreテーブルの列名の一部と見なされます。
   TBLPROPERTIES	外部テーブルのプロパティ設定。以下が含まれます。 tablestore.endpoint：必須。Tablestoreへのアクセスに使用するエンドポイントを指定します。インスタンスに関するエンドポイント情報は、Tablestoreコンソールで確認できます。エンドポイントの詳細については、エンドポイントを参照してください。 tablestore.instance：オプション。Tablestoreインスタンスの名前を指定します。この項目を設定しない場合、インスタンス名はtablestore.endpointの値の最初のフィールドになります。インスタンスの詳細については、インスタンスを参照してください。 tablestore.access_key_id：必須。Alibaba CloudアカウントまたはRAMユーザーのAccessKey IDを指定します。詳細については、AccessKeyペアを取得するを参照してください。 Security Token Service (STS)から取得した一時的なアクセス認証情報を使用してリソースにアクセスする場合は、tablestore.access_key_idを一時的なアクセス認証情報のAccessKey IDに設定します。 tablestore.access_key_secret：必須。Alibaba CloudアカウントまたはRAMユーザーのAccessKeyシークレットを指定します。詳細については、AccessKeyペアを取得するを参照してください。 STSから取得した一時的なアクセス認証情報を使用してリソースにアクセスする場合は、tablestore.access_key_secretを一時的なアクセス認証情報のAccessKeyシークレットに設定します。 tablestore.sts_token：オプション。一時的なアクセス認証情報のセキュリティトークンを指定します。STSから取得した一時的なアクセス認証情報を使用してリソースにアクセスする場合は、この項目を設定する必要があります。詳細については、RAMポリシーを使用してRAMユーザーに権限を付与するを参照してください。 tablestore.table.name：必須。Tablestoreテーブルの名前を指定します。

3. 外部テーブルのデータをクエリします。

   • SELECT * FROM pet;ステートメントを実行して、テーブル内のすべての行をクエリします。

     出力例：

     Bowser  Diane   dog     m       1979-08-31      1995-07-29
     Buffy   Harold  dog     f       1989-05-13      NULL
     Chirpy  Gwen    bird    f       1998-09-11      NULL
     Claws   Gwen    cat     m       1994-03-17      NULL
     Fang    Benny   dog     m       1990-08-27      NULL
     Fluffy  Harold  cat     f       1993-02-04      NULL
     Puffball        Diane   hamster f       1999-03-30      NULL
     Slim    Benny   snake   m       1996-04-29      NULL
     Whistler        Gwen    bird    NULL    1997-12-09      NULL
     Time taken: 5.045 seconds, Fetched 9 row(s)

   • SELECT * FROM pet WHERE birth > "1995-01-01";ステートメントを実行して、birth列の値が1995-01-01より大きい行をクエリします。

     出力例：

     Chirpy  Gwen    bird    f       1998-09-11      NULL
     Puffball        Diane   hamster f       1999-03-30      NULL
     Slim    Benny   snake   m       1996-04-29      NULL
     Whistler        Gwen    bird    NULL    1997-12-09      NULL
     Time taken: 1.41 seconds, Fetched 4 row(s)

HadoopMRを使用してTablestoreテーブルにアクセスする

次のサンプルコードは、HadoopMRを使用してpetデータテーブルの行数に関する統計情報を収集する方法の例を示しています。

• MapperとReducerを構築する

  public class RowCounter {
  public static class RowCounterMapper
  extends Mapper<PrimaryKeyWritable, RowWritable, Text, LongWritable> {
      private final static Text agg = new Text("TOTAL");
      private final static LongWritable one = new LongWritable(1);
  
      // データソースがTablestoreからデータの行を読み取るたびに、mapperのmap()関数が呼び出されます。
      @Override
      public void map(
          PrimaryKeyWritable key, RowWritable value, Context context)
          throws IOException, InterruptedException {
          context.write(agg, one);
      }
  }
  
  public static class IntSumReducer
  extends Reducer<Text,LongWritable,Text,LongWritable> {
  
      @Override
      public void reduce(
          Text key, Iterable<LongWritable> values, Context context)
          throws IOException, InterruptedException {
          long sum = 0;
          for (LongWritable val : values) {
              sum += val.get();
          }
          context.write(key, new LongWritable(sum));
      }
  }
  }
                      

  mapperのmap()関数は、データソースがTablestoreからデータの行を読み取るたびに呼び出されます。PrimaryKeyWritableパラメーターは、行のプライマリキーを指定します。RowWritableパラメーターは、行の内容を指定します。PrimaryKeyWritable.getPrimaryKey()を呼び出して、Tablestore SDK for Javaで定義されているプライマリキーオブジェクトを取得し、RowWritable.getRow()を呼び出して、SDKで定義されている行オブジェクトを取得できます。

• MapperのデータソースとしてTablestoreを指定する

  private static RangeRowQueryCriteria fetchCriteria() {
      RangeRowQueryCriteria res = new RangeRowQueryCriteria("YourTableName");
      res.setMaxVersions(1);
      List<PrimaryKeyColumn> lower = new ArrayList<PrimaryKeyColumn>();
      List<PrimaryKeyColumn> upper = new ArrayList<PrimaryKeyColumn>();
      lower.add(new PrimaryKeyColumn("YourPkeyName", PrimaryKeyValue.INF_MIN));
      upper.add(new PrimaryKeyColumn("YourPkeyName", PrimaryKeyValue.INF_MAX));
      res.setInclusiveStartPrimaryKey(new PrimaryKey(lower));
      res.setExclusiveEndPrimaryKey(new PrimaryKey(upper));
      return res;
  }
  
  public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
      Job job = Job.getInstance(conf, "row count");
      // hadoop-connector.jarをクラスターにデプロイし、hadoop-connector.jarのパスをクラスパスに追加します。
      job.addFileToClassPath(new Path("hadoop-connector.jar"));
      job.setJarByClass(RowCounter.class);
      job.setMapperClass(RowCounterMapper.class);
      job.setCombinerClass(IntSumReducer.class);
      job.setReducerClass(IntSumReducer.class);
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(LongWritable.class);
      job.setInputFormatClass(TableStoreInputFormat.class);
      TableStoreInputFormat.setEndpoint(job, "https://YourInstance.Region.ots.aliyuncs.com/");
      TableStoreInputFormat.setCredential(job, "YourAccessKeyId", "YourAccessKeySecret");
      TableStoreInputFormat.addCriteria(job, fetchCriteria());
      FileOutputFormat.setOutputPath(job, new Path("output"));
      System.exit(job.waitForCompletion(true) ? 0 : 1);
  }                    

  上記の例では、job.setInputFormatClass(TableStoreInputFormat.class)を使用して、データソースとしてTablestoreを指定しています。また、次の操作も実行する必要があります。

  • hadoop-connector.jarをクラスターにデプロイし、hadoop-connector.jarのパスをクラスパスに追加します。パスはhadoop-connector.jarのローカルパスです。パスはaddFileToClassPath()で指定されます。サンプルコードでは、hadoop-connector.jarは現在のパスにあります。

  • Tablestoreへのアクセスに使用するエンドポイントとAccessKeyペアを指定します。TableStoreInputFormat.setEndpoint()を使用してエンドポイントを指定し、TableStoreInputFormat.setCredential()を使用してAccessKeyペアを指定します。

  • Tablestoreテーブルの行数に関する統計情報を記録するテーブルを指定します。

    説明

    • addCriteria()が呼び出されるたびに、Tablestore SDK for Javaで定義されているRangeRowQueryCriteriaオブジェクトがデータソースに追加されます。addCriteria()は複数回呼び出すことができます。RangeRowQueryCriteriaオブジェクトの制限は、Tablestore SDK for JavaのGetRange操作で使用されるRangeRowQueryCriteriaオブジェクトの制限と同じです。

    • RangeRowQueryCriteriaのsetFilter()とaddColumnsToGet()を使用して、Tablestoreサーバー上で不要な行と列を除外できます。これにより、アクセスされるデータ量とコストが削減され、パフォーマンスが向上します。

    • 複数のテーブルに複数のRangeRowQueryCriteriaオブジェクトを追加して、複数の独立したテーブルの結果を組み合わせたユニオンクエリを実行できます。

    • 単一のテーブルに複数のRangeRowQueryCriteriaオブジェクトを追加して、範囲を均等に分割できます。TableStore-Hadoop Connectorは、特定のポリシーに基づいてユーザーが指定した範囲を分割できます。

プログラムを実行する

1. HADOOP_CLASSPATHを指定します。

   HADOOP_CLASSPATH=hadoop-connector.jar bin/hadoop jar row-counter.jar

2. find output -type fコマンドを実行して、出力ディレクトリ内のすべてのファイルをクエリします。

   出力例：

   output/_SUCCESS
   output/part-r-00000
   output/._SUCCESS.crc
   output/.part-r-00000.crc

3. cat output/part-r-00000コマンドを実行して、output/.part-r-00000ファイルの行数に関する統計情報を収集します。

   TOTAL   9

データ型の変換

Tablestoreは、HiveまたはSparkでサポートされているデータ型と部分的に同一のデータ型をサポートしています。

次の表に、Tablestoreでサポートされているデータ型（行）からHiveまたはSparkでサポートされているデータ型（列）への変換を示します。