Tablestore:チュートリアル

データ準備

Tablestore に pet という名前のデータテーブルを作成します。name 列が唯一のプライマリキー列です。以下の表にサンプルデータを示します。

Hive アクセスの例

1. /etc/profile ファイルに HADOOP_HOME と HADOOP_CLASSPATH を追加します。例：

   export HADOOP_HOME=${YourHadoopInstallationDirectory}
   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 を起動し、外部テーブルを作成します。例：

   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.columns.mapping を指定する必要があります。このパラメーターはカンマ区切りの文字列です。カンマの間にスペースを入れないでください。各項目はテーブルの列名であり、項目の順序は外部テーブルのフィールドの順序と一致する必要があります。 説明 Tablestore の列名には空白文字を含めることができます。空白文字は列名の一部と見なされます。
   TBLPROPERTIES	テーブルのプロパティ設定。この設定には次のオプションが含まれます： tablestore.endpoint (必須)：Tablestore にアクセスするために使用されるエンドポイント。Tablestore コンソールでインスタンスのエンドポイントを確認できます。エンドポイントの詳細については、「エンドポイント」をご参照ください。 tablestore.instance (オプション)：Tablestore インスタンスの名前。このパラメーターを空白のままにすると、tablestore.endpoint の最初のセグメントが使用されます。インスタンスの詳細については、「インスタンス」をご参照ください。 tablestore.access_key_id (必須)：Alibaba Cloud アカウントまたは Resource Access Management (RAM) ユーザーの AccessKey ID。詳細については、「AccessKey ペアの取得」をご参照ください。Security Token Service (STS) を使用してリソースに一時的にアクセスする場合は、このパラメーターを一時的なアクセス認証情報の AccessKey ID に設定します。 tablestore.access_key_secret (必須)：Alibaba Cloud アカウントまたは RAM ユーザーの AccessKey シークレット。詳細については、「AccessKey ペアの取得」をご参照ください。STS を使用してリソースに一時的にアクセスする場合は、このパラメーターを一時的なアクセス認証情報の 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 アクセスの例

次の例は、HadoopMR プログラムを使用して pet データテーブルの行数をカウントする方法を示しています。

• マッパーとレデューサーを構築します。

  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);
  
      @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));
      }
  }
  }

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

• Tablestore をマッパーのデータソースとして設定します。

  private static RangeRowQueryCriteria fetchCriteria() {
      RangeRowQueryCriteria res = new RangeRowQueryCriteria("pet");
      res.setMaxVersions(1);
      List<PrimaryKeyColumn> lower = new ArrayList<PrimaryKeyColumn>();
      List<PrimaryKeyColumn> upper = new ArrayList<PrimaryKeyColumn>();
      lower.add(new PrimaryKeyColumn("name", PrimaryKeyValue.INF_MIN));
      upper.add(new PrimaryKeyColumn("name", 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");
      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 をクラスターにデプロイし、クラスパスに追加します。addFileToClassPath() を使用して hadoop-connector.jar のローカルパスを指定します。この例では、hadoop-connector.jar が現在のパスにあることを前提としています。

  • Tablestore にアクセスするために必要なエンドポイントと認証情報を指定します。TableStoreInputFormat.setEndpoint() および TableStoreInputFormat.setCredential() を使用して、Tablestore にアクセスするために必要なエンドポイントと AccessKey 情報を指定します。

  • カウントするテーブルを指定します。

  説明

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

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

  • 複数のテーブルに対して UNION 演算を実行するには、テーブルに対応する複数の RangeRowQueryCriteria オブジェクトを追加します。

  • データをより均等に分割するには、同じテーブルに対して複数の RangeRowQueryCriteria オブジェクトを追加します。Tablestore-Hadoop コネクタは、指定された範囲を特定のポリシーに基づいてより小さな範囲に分割します。

プログラム実行の例

1. HADOOP_CLASSPATH を設定します。

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

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

   コマンドは次の結果を返します：

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

3. cat output/part-r-00000 コマンドを実行して、実行結果の行数を表示します。

   TOTAL   9

型変換に関する注意事項

Tablestore がサポートするデータ型は、Hive または Spark がサポートするデータ型と完全には一致しません。

以下の表は、Tablestore のデータ型 (行) から Hive または Spark のデータ型 (列) への変換のサポート状況を示しています。