AdalbertMemSQL commented on code in PR #23535: URL: https://github.com/apache/beam/pull/23535#discussion_r1024355506
########## sdks/java/io/singlestore/src/main/java/org/apache/beam/sdk/io/singlestore/SingleStoreIO.java: ########## @@ -0,0 +1,955 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.singlestore; + +import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument; +import static org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull; + +import com.google.auto.value.AutoValue; +import java.io.IOException; +import java.io.InputStream; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.io.Serializable; +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.List; +import javax.sql.DataSource; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.coders.ListCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.io.range.OffsetRange; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.Filter; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.Reshuffle; +import org.apache.beam.sdk.transforms.SerializableFunctions; +import org.apache.beam.sdk.transforms.View; +import org.apache.beam.sdk.transforms.display.DisplayData; +import org.apache.beam.sdk.transforms.splittabledofn.RestrictionTracker; +import org.apache.beam.sdk.transforms.windowing.GlobalWindow; +import org.apache.beam.sdk.util.Preconditions; +import org.apache.beam.sdk.values.PBegin; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollectionView; +import org.apache.commons.dbcp2.BasicDataSource; +import org.apache.commons.dbcp2.DelegatingStatement; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.joda.time.Instant; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * IO to read and write data on SingleStoreDB. + * + * <h3>Reading from SingleStoreDB datasource</h3> + * + * <p>SingleStoreIO source returns a bounded collection of {@code T} as a {@code PCollection<T>}. T + * is the type returned by the provided {@link RowMapper}. + * + * <p>To configure the SingleStoreDB source, you have to provide a {@link DataSourceConfiguration} + * using {@link DataSourceConfiguration#create(String)}(endpoint). Optionally, {@link + * DataSourceConfiguration#withUsername(String)} and {@link + * DataSourceConfiguration#withPassword(String)} allows you to define username and password. + * + * <p>For example: + * + * <pre>{@code + * 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)); + * } + * }) + * ); + * }</pre> + * + * <p>Query parameters can be configured using a user-provided {@link StatementPreparator}. For + * example: + * + * <pre>{@code + * 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)); + * } + * }) + * ); + * }</pre> + * + * <h4>Parallel reading from a SingleStoreDB datasource</h4> + * + * <p>SingleStoreIO supports partitioned reading of all data from a table. To enable this, use + * {@link SingleStoreIO#readWithPartitions()}. This way of data reading is preferred because of the + * performance reasons. + * + * <p>The following example shows usage of {@link SingleStoreIO#readWithPartitions()} + * + * <pre>{@code + * 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)); + * } + * }) + * ); + * }</pre> + * + * <h3>Writing to SingleStoreDB datasource</h3> + * + * <p>SingleStoreIO supports writing records into a database. It writes a {@link PCollection} to the + * database by converting data to CSV and sending it to the database with LOAD DATA query. + * + * <p>Like the source, to configure the sink, you have to provide a {@link DataSourceConfiguration}. + * + * <pre>{@code + * 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; + * } + * }) + * ); + * }</pre> + */ +final class SingleStoreIO { Review Comment: Yea, sorry -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
