开放存储SDK示例-Java SDK

SplitOptions

• 参数定义

  SplitOptions对象参数定义如下：

  public class SplitOptions {
  
      public static SplitOptions.Builder newBuilder() {
          return new Builder();
      }
  
      public static class Builder {
  
        public SplitOptions.Builder SplitByByteSize(long splitByteSize);
    
        public SplitOptions.Builder SplitByRowOffset();
    
        public SplitOptions.Builder withCrossPartition(boolean crossPartition);
    
        public SplitOptions.Builder withMaxFileNum(int splitMaxFileNum);
    
        public SplitOptions build();
      }
  }

• 参数说明

  • SplitByByteSize(long splitByteSize)

    按照指定的参数splitByteSize切分数据，服务端返回的单个数据分片大小不超过splitByteSize（单位为：byte）。

    • 自定义切分数据值时不能小于10*1024*1024（10 MB）。

    • 若未使用SplitByByteSize(long splitByteSize)自定义切分数据值，系统将默认按照256*1024*1024（256 MB）切分数据。

  • SplitByRowOffset()

    按照行切分数据，允许客户端从指定的行索引读取数据。

  • withCrossPartition(boolean crossPartition)

    是否允许单个数据分片包含多个数据分区。crossPartition参数取值如下：

    • true（默认值）：允许单个数据分片包含多个数据分区。

    • false：不允许。

  • withMaxFileNum(int splitMaxFileNum)

    在表文件数量较多的情况下，可以通过指定单个数据分片中最大包含的物理文件数量来产生更多的数据分片。

    默认不限制单个数据分片中包含的物理文件数量。

  • build()：创建SplitOptions对象。

• 使用示例

  // 1. 按照SplitSize切分数据，设置SplitSize为256MB
  
  SplitOptions splitOptionsByteSize = 
        SplitOptions.newBuilder().SplitByByteSize(256 * 1024L * 1024L).build()
  
  // 2. 按照RowOffset切分数据
  
  SplitOptions splitOptionsCount = 
        SplitOptions.newBuilder().SplitByRowOffset().build()
  
  //3. 指定单个Split包含的最大文件数为1
  
  SplitOptions splitOptionsCount = 
        SplitOptions.newBuilder().SplitByRowOffset().withMaxFileNum(1).build()

ArrowOptions

• 参数定义

  ArrowOptions对象参数定义如下：

  public class ArrowOptions {
      
      public static Builder newBuilder() {
          return new Builder();
      }
  
      public static class Builder {
  
          public Builder withTimestampUnit(TimestampUnit unit);
  
          public Builder withDatetimeUnit(TimestampUnit unit);
  
          public ArrowOptions build();
      }
  
      public enum TimestampUnit {
          SECOND,
          MILLI,
          MICRO,
          NANO;
      }
  }

• 参数说明

  • TimestampUnit

    用于指定Timestamp和Datetime数据类型的单位，取值如下：

    • SECOND：秒（s）。

    • MILLI：毫秒（ms）。

    • MICRO：微秒（μs）。

    • NANO：纳秒（ns）。

  • withTimestampUnit(TimestampUnit unit)

    指定Timestamp数据类型的单位，默认单位为NANO。

  • withDatetimeUnit(TimestampUnit unit)

    指定Datetime数据类型的单位，默认单位为MILLI。

• 使用示例

  ArrowOptions options = ArrowOptions.newBuilder()
            .withDatetimeUnit(ArrowOptions.TimestampUnit.MILLI)
            .withTimestampUnit(ArrowOptions.TimestampUnit.NANO)
            .build()

Predicate

• 参数定义

  SplitOptions对象参数定义如下：

  // 1. 二元运算
  
  public class BinaryPredicate extends Predicate {
  
    public enum Operator {
      /**
       * 二元运算操作符
       */
      EQUALS("="),
      NOT_EQUALS("!="),
      GREATER_THAN(">"),
      LESS_THAN("<"),
      GREATER_THAN_OR_EQUAL(">="),
      LESS_THAN_OR_EQUAL("<=");
     
    }
  
    public BinaryPredicate(Operator operator, Serializable leftOperand, Serializable rightOperand);
  
    public static BinaryPredicate equals(Serializable leftOperand, Serializable rightOperand);
  
    public static BinaryPredicate notEquals(Serializable leftOperand, Serializable rightOperand);
  
    public static BinaryPredicate greaterThan(Serializable leftOperand, Serializable rightOperand);
  
    public static BinaryPredicate lessThan(Serializable leftOperand, Serializable rightOperand);
  
    public static BinaryPredicate greaterThanOrEqual(Serializable leftOperand,
                                                     Serializable rightOperand);
  
    public static BinaryPredicate lessThanOrEqual(Serializable leftOperand,
                                                  Serializable rightOperand);
  }
  
  // 2. 一元运算
  public class UnaryPredicate extends Predicate {
  
    public enum Operator {
      /**
       * 一元运算操作符
       */
      IS_NULL("is null"),
      NOT_NULL("is not null");
    }
  
    public static UnaryPredicate isNull(Serializable operand);
    public static UnaryPredicate notNull(Serializable operand);
  }
  
  ### 3. IN and NOT IN
  public class InPredicate extends Predicate {
  
    public enum Operator {
      /**
       * IN and NOT IN operators for set membership check
       */
      IN("in"),
      NOT_IN("not in");
    }
  
    public InPredicate(Operator operator, Serializable operand, List<Serializable> set);
  
    public static InPredicate in(Serializable operand, List<Serializable> set);
  
    public static InPredicate notIn(Serializable operand, List<Serializable> set);
  }
  
  // 4. 列名
  public class Attribute extends Predicate {
  
    public Attribute(Object value);
      
    public static Attribute of(Object value);
  }
  
  // 5. 常量
  public class Constant extends Predicate {
  
    public Constant(Object value);
  
    public static Constant of(Object value);
  }
  
  // 6. 组合运算
  public class CompoundPredicate extends Predicate {
  
    public enum Operator {
      /**
       * 复合谓词运算符
       */
      AND("and"),
      OR("or"),
      NOT("not");
    }
  
    public CompoundPredicate(Operator logicalOperator, List<Predicate> predicates);
  
    public static CompoundPredicate and(Predicate... predicates);
    
    public static CompoundPredicate or(Predicate... predicates);
  
    public static CompoundPredicate not(Predicate predicates);
  
    public void addPredicate(Predicate predicate);
  }
  
  // 7. 原始谓词 RawPredicate
  // 如现有方法不满足需求，可以根据 SQL 语法自行组装谓词部分
    public class RawPredicate extends Predicate {
        public RawPredicate(String rawExpr);
        public static RawPredicate of(String rawExpr);
    }

• 使用示例

  // 1. c1 > 20000 and c2 < 100000
  BinaryPredicate c1 = new BinaryPredicate(BinaryPredicate.Operator.GREATER_THAN, Attribute.of("c1"), Constant.of(20000));
  BinaryPredicate c2 = new BinaryPredicate(BinaryPredicate.Operator.LESS_THAN, Attribute.of("c2"), Constant.of(100000));
  CompoundPredicate predicate =
          new CompoundPredicate(CompoundPredicate.Operator.AND, ImmutableList.of(c1, c2));
  
  // 2. c1 is not null
  Predicate predicate = new UnaryPredicate(UnaryPredicate.Operator.NOT_NULL,  Attribute.of("c1"));
  
    
  // 3. c1 in (1, 10001)
  Predicate predicate =
          new InPredicate(InPredicate.Operator.IN,  Attribute.of("c1"), ImmutableList.of(Constant.of(1), Constant.of(10001)));
  
  // 4. 使用 RawPredicate 自行组装谓词（支持全类型）
    Predicate predicate = RawPredicate.of("c1 > 20000 and c2 < 100000");

EnvironmentSettings

• 参数定义

  接口EnvironmentSettings定义如下：

  public class EnvironmentSettings {
  
      public static Builder newBuilder() {
          return new Builder();
      }
  
      public static class Builder {
  
          public Builder withDefaultProject(String projectName);
  
          public Builder withDefaultSchema(String schema);
  
          public Builder withServiceEndpoint(String endPoint);
  
          public Builder withTunnelEndpoint(String tunnelEndPoint);
  
          public Builder withQuotaName(String quotaName);
  
          public Builder withCredentials(Credentials credentials);
  
          public Builder withRestOptions(RestOptions restOptions);
  
          public EnvironmentSettings build();
      }
  }

• 参数说明

  • withDefaultProject(String projectName)

    设置当前的项目名称。projectName参数为MaxCompute项目名称。登录MaxCompute控制台，在左上角选择地域。左侧导航栏，选择管理配置 > 项目管理，可以查看具体的MaxCompute项目名称。

  • withDefaultSchema(String schema)

    设置为当前默认的schema。schema参数为MaxCompute Schema名称，关于Schema详情，请参见Schema操作。

  • withServiceEndpoint(String endPoint)

    设置当前服务链接地址Endpoint。

  • withTunnelEndpoint(String tunnelEndPoint)

    设置当前服务链接地址TunnelEndpoint。

  • withQuotaName(String quotaName)

    指定当前使用的Quota名称。

    访问MaxCompute支持独享数据传输服务资源组（包年包月）和开放存储（按量计费）两种资源，获取Quota名称的方式分别如下：

    • 独享数据传输服务资源组

      登录MaxCompute控制台，在左上角选择地域。

      在左侧导航栏，选择管理配置 > 配额（Quota）管理 。

      查看可使用的Quota列表。具体操作，请参见计算资源-Quota管理。

    • 登录MaxCompute控制台，在左上角选择地域。

      在左侧导航栏，选择管理配置 > 租户管理 。

      在租户属性页签，打开开放存储(Storage API)开关。

  • withCredentials(Credentials credentials)

    指定当前认证信息，详情请参见Credentials。

Credentials

• 对象定义

  public class Credentials {
  
      public static Builder newBuilder() {
          return new Builder();
      }
  
      public static class Builder {
  
          public Builder withAccount(Account account);
  
          public Builder withAppAccount(AppAccount appAccount);
  
          public Builder withAppStsAccount(AppStsAccount appStsAccount);
  
          public Credentials build();
      }
  
  }

• 参数说明

  • withAccount(Account account)

    指定Odps Account对象。

  • withAppAccount(AppAccount appAccount)

    指定Odps appAccount对象。

  • withAppStsAccount(AppStsAccount appStsAccount)

    指定Odps appStsAccount对象。

  • withRestOptions(RestOptions restOptions)

    指定当前访问网络的配置，RestOptions定义如下：

    public class RestOptions implements Serializable {
    
        public static Builder newBuilder() {
            return new RestOptions.Builder();
        }
    
        public static class Builder {
            public Builder witUserAgent(String userAgent);
            public Builder withConnectTimeout(int connectTimeout);
            public Builder withReadTimeout(int readTimeout);
            public RestOptions build();
        }
    }

    • witUserAgent(String userAgent)：指定当前userAgent信息。

    • withConnectTimeout(int connectTimeout)：指定当前底层网络建立超时时间，默认为10秒（s）。

    • withReadTimeout(int readTimeout)：指定当前底层网络连接超时时间，120秒（s）。

• DataSchema定义如下：

  public class DataSchema implements Serializable {
      
      List<Column> getColumns();
  
      List<String> getPartitionKeys();
  
      List<String> getColumnNames();
  
      List<TypeInfo> getColumnDataTypes();
  
      Optional<Column> getColumn(int columnIndex);
  
      Optional<Column> getColumn(String columnName);
  
  }

• 参数说明

  • getColumns()：获取要读取的表和分区的Column信息。

  • getPartitionKeys()：获取要读取的分区Column名称。

  • getColumnNames()：获取要读取的表和分区的Column名称。

  • getColumnDataTypes()：获取要读取的表和分区的Column信息。

  • getColumn(int columnIndex)：根据索引获取Column对象，若索引超出当前Column的范围，则返回为空。

  • getColumn(String columnName)：根据参数columnName获取Column对象，若当前表的Column不包含columnName，则返回为空。

• InputSplitAssigner定义如下：

  public interface InputSplitAssigner {
  
      int getSplitsCount();
  
      long getTotalRowCount();
  
      InputSplit getSplit(int index);
  
      InputSplit getSplitByRowOffset(long startIndex, long numRecord);
  }

• 参数说明

  • getSplitsCount()：获取会话包含的数据分片数量。

    说明

    当SplitOptions为SplitByByteSize时，该接口返回值大于等于0。

  • getTotalRowCount()：获取会话包含的数据行数。

    说明

    当SplitOptions为SplitByByteSize时，该接口返回值大于等于0

  • getSplit(int index)：根据指定的数据分片参数Index获取对应的InputSplit，参数index取值范围为：[0,SplitsCount-1]。

  • getSplitByRowOffset(long startIndex, long numRecord)：获取对应的InputSplit。参数说明如下：

  • startIndex：指定InputSplit读取的数据行起始索引，取值范围为[0,RecordCount-1]。

  • numRecord：指定InputSplit读取的数据行数。

• 使用示例

  // 1. 若SplitOptions为SplitByByteSize
  
  TableBatchReadSession scan = ...;
  InputSplitAssigner assigner = scan.getInputSplitAssigner();
  int splitCount = assigner.getSplitsCount();
  for (int k = 0; k < splitCount; k++) {
      InputSplit split = assigner.getSplit(k);
      ...
  }
  
  // 2. 若SplitOptions为SplitByRowOffset
  TableBatchReadSession scan = ...;
  InputSplitAssigner assigner = scan.getInputSplitAssigner();
  long rowCount = assigner.getTotalRowCount();
  long recordsPerSplit = 10000;
  for (long offset = 0; offset < numRecords; offset += recordsPerSplit) {
      recordsPerSplit = Math.min(recordsPerSplit, numRecords - offset);
      InputSplit split = assigner.getSplitByRowOffset(offset, recordsPerSplit);
      ...
  }

• ReaderOptions定义如下：

  public class ReaderOptions {
      
      public static ReaderOptions.Builder newBuilder() {
          return new Builder();
      }
  
      public static class Builder {
  
          public Builder withMaxBatchRowCount(int maxBatchRowCount);
  
          public Builder withMaxBatchRawSize(long batchRawSize);
  
          public Builder withCompressionCodec(CompressionCodec codec);
  
          public Builder withBufferAllocator(BufferAllocator allocator);
  
          public Builder withReuseBatch(boolean reuseBatch);
  
          public Builder withSettings(EnvironmentSettings settings);
  
          public ReaderOptions build();
      }
  
  }

• 参数说明

  • withMaxBatchRowCount(int maxBatchRowCount)

    指定服务端返回的每批次数据中最大行数。参数maxBatchRowCount默认最大值为4096。

  • withMaxBatchRawSize(long batchRawSize)

    指定服务端返回的每个批次数据包含的最大原始字节数。

  • withCompressionCodec(CompressionCodec codec)

    指定数据压缩类型，目前只支持ZSTD和LZ4_FRAME压缩类型。

    说明

    • 在直接传输大量未加压缩的Arrow数据时，由于网络带宽的限制，可能会显著增加数据传输时间。

    • 若未指定压缩类型，默认不压缩数据。

  • withBufferAllocator(BufferAllocator allocator)

    指定读取Arrow数据的内存分配器。

  • withReuseBatch(boolean reuseBatch)

    指定ArrowBatch内存是否可复用。reuseBatch参数取值如下：

    • true（默认值）：ArrowBatch内存可复用。

    • false：ArrowBatch内存不可复用。

  • withSettings(EnvironmentSettings settings)

    指定运行环境信息。