Package org.apache.beam.sdk.io.singlestore

Class SingleStoreIO

java.lang.Object

org.apache.beam.sdk.io.singlestore.SingleStoreIO

public class SingleStoreIO
extends Object

IO to read and write data on SingleStoreDB.

Reading from SingleStoreDB datasource

SingleStoreIO source returns a bounded collection of T as a PCollection<T>. T is the type returned by the provided SingleStoreIO.RowMapper.

To configure the SingleStoreDB source, you have to provide a SingleStoreIO.DataSourceConfiguration using SingleStoreIO.DataSourceConfiguration.create(String)(endpoint). Optionally, SingleStoreIO.DataSourceConfiguration.withUsername(String) and SingleStoreIO.DataSourceConfiguration.withPassword(String) allows you to define username and password.

For example:

 pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
   .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
        .withUsername("username")
        .withPassword("password"))
   .withQuery("select id, name from Person")
   .withRowMapper(new RowMapper<KV<Integer, String>>() {
     public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
       return KV.of(resultSet.getInt(1), resultSet.getString(2));
     }
   })
 );
 

Query parameters can be configured using a user-provided SingleStoreIO.StatementPreparator. For example:

 pipeline.apply(SingleStoreIO.<KV<Integer, String>>read()
   .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306"))
   .withQuery("select id,name from Person where name = ?")
   .withStatementPreparator(new StatementPreparator() {
     public void setParameters(PreparedStatement preparedStatement) throws Exception {
       preparedStatement.setString(1, "Darwin");
     }
   })
   .withRowMapper(new RowMapper<KV<Integer, String>>() {
     public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
       return KV.of(resultSet.getInt(1), resultSet.getString(2));
     }
   })
 );
 

Parallel reading from a SingleStoreDB datasource

SingleStoreIO supports partitioned reading of all data from a table. To enable this, use readWithPartitions(). This way of data reading is preferred because of the performance reasons.

The following example shows usage of readWithPartitions()

 pipeline.apply(SingleStoreIO.<Row>readWithPartitions()
  .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
       .withUsername("username")
       .withPassword("password"))
  .withTable("Person")
  .withRowMapper(new RowMapper<KV<Integer, String>>() {
    public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
      return KV.of(resultSet.getInt(1), resultSet.getString(2));
    }
  })
 );
 

Writing to SingleStoreDB datasource

SingleStoreIO supports writing records into a database. It writes a PCollection to the database by converting data to CSV and sending it to the database with LOAD DATA query.

Like the source, to configure the sink, you have to provide a SingleStoreIO.DataSourceConfiguration.

 pipeline
   .apply(...)
   .apply(SingleStoreIO.<KV<Integer, String>>write()
      .withDataSourceConfiguration(DataSourceConfiguration.create("hostname:3306")
          .withUsername("username")
          .withPassword("password"))
      .withStatement("insert into Person values(?, ?)")
      .withUserDataMapper(new UserDataMapper<KV<Integer, String>>() {
        @Override
        public List<String> mapRow(KV<Integer, String> element) {
          List<String> res = new ArrayList<>();
          res.add(element.getKey().toString());
          res.add(element.getValue());
          return res;
        }
      })
    );
 

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	SingleStoreIO.DataSourceConfiguration	A POJO describing a SingleStoreDB DataSource by providing all properties needed to create it.
static class	SingleStoreIO.Read<T>	A PTransform for reading data from SingleStoreDB.
static class	SingleStoreIO.ReadWithPartitions<T>	A PTransform for reading data from SingleStoreDB.
static interface	SingleStoreIO.RowMapper<T>	An interface used by SingleStoreIO.Read and SingleStoreIO.ReadWithPartitions for converting each row of the ResultSet into an element of the resulting PCollection.
static interface	SingleStoreIO.RowMapperWithCoder<T>	A RowMapper that provides a Coder for resulting PCollection.
static interface	SingleStoreIO.RowMapperWithInit<T>	A RowMapper that requires initialization.
static interface	SingleStoreIO.StatementPreparator	An interface used by the SingleStoreIO SingleStoreIO.Read to set the parameters of the PreparedStatement.
static interface	SingleStoreIO.UserDataMapper<T>	An interface used by the SingleStoreIO SingleStoreIO.Write to map a data from each element of PCollection to a List of Strings.
static class	SingleStoreIO.Write<T>	A PTransform for writing data to SingleStoreDB.

Constructor Summary

Constructors

Constructor	Description
SingleStoreIO()

Method Summary

Modifier and Type	Method	Description
static <T> SingleStoreIO.Read<T>	read()	Read data from a SingleStoreDB datasource.
static SingleStoreIO.Read<Row>	readRows()	Read Beam Rows from a SingleStoreDB datasource.
static <T> SingleStoreIO.ReadWithPartitions<T>	readWithPartitions()	Like read(), but executes multiple instances of the query on the same table for each database partition.
static SingleStoreIO.ReadWithPartitions<Row>	readWithPartitionsRows()	Like readRows(), but executes multiple instances of the query on the same table for each database partition.
static <T> SingleStoreIO.Write<T>	write()	Write data to a SingleStoreDB datasource.
static SingleStoreIO.Write<Row>	writeRows()	Write Beam Rows to a SingleStoreDB datasource.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

SingleStoreIO

public SingleStoreIO()

Method Details

read

public static <T> SingleStoreIO.Read<T> read()

Read data from a SingleStoreDB datasource.

Type Parameters:

T - Type of the data to be read.

readRows

public static SingleStoreIO.Read<Row> readRows()

Read Beam Rows from a SingleStoreDB datasource.

readWithPartitions

public static <T>
SingleStoreIO.ReadWithPartitions<T> readWithPartitions()

Like read(), but executes multiple instances of the query on the same table for each database partition.

Type Parameters:

T - Type of the data to be read.

readWithPartitionsRows

public static SingleStoreIO.ReadWithPartitions<Row> readWithPartitionsRows()

Like readRows(), but executes multiple instances of the query on the same table for each database partition.

write

public static <T> SingleStoreIO.Write<T> write()

Write data to a SingleStoreDB datasource.

Type Parameters:

T - Type of the data to be written.

writeRows

public static SingleStoreIO.Write<Row> writeRows()

Write Beam Rows to a SingleStoreDB datasource.