MaxCompute：開放儲存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支援獨享Data Transmission Service資源群組（訂用帳戶）資源，擷取Quota名稱的方式分別如下：

    • 獨享Data Transmission Service資源群組

      登入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資料的記憶體 Clerk。

  • withReuseBatch(boolean reuseBatch)

    指定ArrowBatch記憶體是否可複用。reuseBatch參數取值如下：

    • true（預設值）：ArrowBatch記憶體可複用。

    • false：ArrowBatch記憶體不可複用。

  • withSettings(EnvironmentSettings settings)

    指定運行環境資訊。