内置 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 文档

身份验证

DataSource 配置是为配置 SingleStoreIO 连接属性所必需的。

创建 DataSource 配置

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

其中参数可以是

• .create(endpoint)
  • SingleStoreDB 的主机名或 IP 地址，格式为 host:[port]（端口可选）。
  • 必需参数。
  • 示例：.create("myHost:3306")。
• .withUsername(username)
  • SingleStoreDB 用户名。
  • 默认值 - root。
  • 示例：.withUsername("USERNAME")。
• .withPassword(password)
  • SingleStoreDB 用户的密码。
  • 默认值 - 空字符串。
  • 示例：.withPassword("PASSWORD")。
• .withDatabase(database)
  • 要使用的 SingleStoreDB 数据库的名称。
  • 示例：.withDatabase("MY_DATABASE")。
• .withConnectionProperties(connectionProperties)
  • JDBC 驱动程序使用的属性列表。
  • 格式为“key1=value1;key2=value2;…”。
  • 可以在 此处 找到支持的属性完整列表。
  • 示例：.withConnectionProperties("connectTimeout=30000;useServerPrepStmts=FALSE")。

注意 - .withDatabase(...) 对于 .readWithPartitions() 来说是必需的。

从 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(table)
  • 要从中读取数据的表。
  • 示例：.withTable("MY_TABLE")。
• .withQuery(query)
  • 要执行的 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(table)
  • 要从中读取数据的表。
  • 示例：.withTable("MY_TABLE")。
• .withQuery(query)
  • 要执行的 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(table)
  • 应该保存数据的表。
  • 必需参数。
  • 示例：.withTable("MY_TABLE")。
• .withBatchSize(batchSize)
  • 一个 LOAD DATA 查询加载的行数。
  • 默认值 - 100000。
  • 示例：.withBatchSize(100000)。
• .withUserDataMapper(userDataMapper)
  • UserDataMapper 对象。
  • 必需参数。

UserDataMapper

UserDataMapper 是必需的，用于将数据从 PCollection 映射到 String 值数组，然后 write() 操作保存数据。例如

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