Spark + TableStore

Spark接入TableStore

• 准备一张数据表

创建一张表 pet，其中 name 为主键。

• 以下示例演示了如何在 Spark 中消费 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) {
      SparkConf sparkConf = new SparkConf().setAppName("RowCounter");
      JavaSparkContext sc = new JavaSparkContext(sparkConf);
      Configuration hadoopConf = new Configuration();
      JavaSparkContext sc = null;
      try {
          sc = new JavaSparkContext(sparkConf);
          Configuration hadoopConf = new Configuration();
          TableStore.setCredential(
                  hadoopConf,
                  new Credential(accessKeyId, accessKeySecret, securityToken));
          Endpoint ep = new Endpoint(endpoint, instance);
          TableStore.setEndpoint(hadoopConf, ep);
          TableStoreInputFormat.addCriteria(hadoopConf, fetchCriteria());
          JavaPairRDD<PrimaryKeyWritable, RowWritable> rdd = sc.newAPIHadoopRDD(
                  hadoopConf, TableStoreInputFormat.class,
                  PrimaryKeyWritable.class, RowWritable.class);
          System.out.println(
              new Formatter().format("TOTAL: %d", rdd.count()).toString());
      } finally {
          if (sc != null) {
              sc.close();
          }
      }
  }

附录

完整示例代码请参见：

• Spark 接入 TableStore