內建 I/O 轉換

SingleStoreDB I/O

管道選項和有關使用和執行 SingleStoreDB I/O 的一般資訊。

開始之前

若要使用 SingleStoreDB I/O，請將 Maven 成品相依性新增至您的 pom.xml 檔案。

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-singlestore</artifactId>
    <version>2.60.0</version>
</dependency>

其他資源

• SingleStoreIO 原始碼
• SingleStoreIO Javadoc
• SingleStore 文件

驗證

設定 SingleStoreIO 連線屬性需要 DataSource 設定。

建立 DataSource 設定

SingleStoreIO.DataSourceConfiguration
    .create("myHost:3306")
    .withDatabase("db")
    .withConnectionProperties("connectTimeout=30000;useServerPrepStmts=FALSE")
    .withPassword("password")
    .withUsername("admin");

參數可以是

• .create(端點)
  • SingleStoreDB 的主機名稱或 IP 位址，格式為 host:[port] (port 是選用的)。
  • 必要參數。
  • 範例：.create("myHost:3306")。
• .withUsername(使用者名稱)
  • SingleStoreDB 使用者名稱。
  • 預設值 - root。
  • 範例：.withUsername("USERNAME")。
• .withPassword(密碼)
  • SingleStoreDB 使用者的密碼。
  • 預設值 - 空字串。
  • 範例：.withPassword("PASSWORD")。
• .withDatabase(資料庫)
  • 要使用的 SingleStoreDB 資料庫名稱。
  • 範例：.withDatabase("MY_DATABASE")。
• .withConnectionProperties(連線屬性)
  • JDBC 驅動程式使用的屬性清單。
  • 格式為「key1=value1;key2=value2;…」。
  • 完整的支援屬性清單可在此處找到。
  • 範例：.withConnectionProperties("connectTimeout=30000;useServerPrepStmts=FALSE")。

注意 - .readWithPartitions() 需要 .withDatabase(...)。

從 SingleStoreDB 讀取

SingleStoreIO 的其中一個功能是從 SingleStoreDB 資料表讀取。SingleStoreIO 支援兩種讀取類型

• 循序資料讀取 (.read())
• 平行資料讀取 (.readWithPartitions())

在許多情況下，由於效能原因，平行資料讀取優於循序資料讀取。

循序資料讀取

基本的 .read() 操作使用方式如下

PCollection<USER_DATA_TYPE> items = pipeline.apply(
    SingleStoreIO.<USER_DATA_TYPE>read()
        .withDataSourceConfiguration(dc)
        .withTable("MY_TABLE") // or .withQuery("QUERY")
        .withStatementPreparator(statementPreparator)
        .withOutputParallelization(true)
        .withRowMapper(mapper)
);

參數可以是

• .withDataSourceConfiguration(dataSourceConfiguration)
  • 具有建立資料庫連線所需的所有資訊的 DataSourceConfiguration 物件。如需詳細資訊，請參閱驗證。
  • 必要參數。
• .withTable(資料表)
  • 要從中讀取資料的資料表。
  • 範例：.withTable("MY_TABLE")。
• .withQuery(查詢)
  • 要執行的 SQL 查詢。
  • 範例：.withTable("SELECT * FROM MY_TABLE")。
• .withStatementPreparator(statementPreparator)
  • StatementPreparator 物件。
• .withRowMapper(rowMapper)
  • RowMapper 物件。
  • 必要參數。
• .withOutputParallelization(outputParallelization)
  • 布林值，指出是否重新混洗結果。
  • 預設值 - true。
  • 範例：.withOutputParallelization(true)。

注意 - 必須使用 .withTable(...) 或 .withQuery(...) 其中之一。

平行資料讀取

基本的 .readWithPartitions() 操作使用方式如下

PCollection<USER_DATA_TYPE> items = pipeline.apply(
    SingleStoreIO.<USER_DATA_TYPE>readWithPartitions()
        .withDataSourceConfiguration(dc)
        .withTable("MY_TABLE") // or .withQuery("QUERY")
        .withRowMapper(mapper)
);

參數可以是

• .withDataSourceConfiguration(dataSourceConfiguration)
  • 具有建立資料庫連線所需的所有資訊的 DataSourceConfiguration 物件。如需詳細資訊，請參閱DataSource 設定。
  • 必要參數。
• .withTable(資料表)
  • 要從中讀取資料的資料表。
  • 範例：.withTable("MY_TABLE")。
• .withQuery(查詢)
  • 要執行的 SQL 查詢。
  • 範例：.withTable("SELECT * FROM MY_TABLE")。
• .withRowMapper(rowMapper)
  • RowMapper 物件。
  • 必要參數。

注意 - 必須使用 .withTable(...) 或 .withQuery(...) 其中之一。

StatementPreparator

StatementPreparator 由 read() 用來設定 PreparedStatement 的參數。例如

public static class MyStatmentPreparator implements SingleStoreIO.StatementPreparator {
    @Override
    public void setParameters(PreparedStatement preparedStatement) throws Exception {
        preparedStatement.setInt(1, 10);
    }
}

RowMapper

RowMapper 由 read() 和 readWithPartitions() 用於將 ResultSet 的每一列轉換為產生的 PCollection 的元素。例如

public static class MyRowMapper implements SingleStoreIO.RowMapper<MyRow> {
    @Override
    public MyRow mapRow(ResultSet resultSet) throws Exception {
        return MyRow.create(resultSet.getInt(1), resultSet.getString(2));
    }
}

寫入 SingleStoreDB 資料表

SingleStoreIO 的其中一個功能是寫入 SingleStoreDB 資料表。此轉換使您能夠將使用者的 PCollection 傳送到您的 SingleStoreDB 資料庫。它會傳回每一批元素寫入的列數。

基本的 .write() 操作使用方式如下

data.apply(
    SingleStoreIO.<USER_DATA_TYPE>write()
        .withDataSourceConfiguration(dc)
        .withTable("MY_TABLE")
        .withUserDataMapper(mapper)
        .withBatchSize(100000)
);

參數可以是

• .withDataSourceConfiguration(dataSourceConfiguration)
  • 具有建立資料庫連線所需的所有資訊的 DataSourceConfiguration 物件。如需詳細資訊，請參閱DataSource 設定。
  • 必要參數。
• .withTable(資料表)
  • 應該在其中儲存資料的資料表。
  • 必要參數。
  • 範例：.withTable("MY_TABLE")。
• .withBatchSize(批次大小)
  • 一個 LOAD DATA 查詢載入的列數。
  • 預設值 - 100000。
  • 範例：.withBatchSize(100000)。
• .withUserDataMapper(userDataMapper)
  • UserDataMapper 物件。
  • 必要參數。

UserDataMapper

在 write() 操作儲存資料之前，需要 UserDataMapper 將資料從 PCollection 對應到 String 值的陣列。例如

public static class MyRowDataMapper implements SingleStoreIO.UserDataMapper<MyRow> {
    @Override
    public List<String> mapRow(MyRow element) {
        List<String> res = new ArrayList<>();
        res.add(element.id().toString());
        res.add(element.name());
        return res;
    }
}