在程序中通过云原生数据仓库 AnalyticDB MySQL 版Client高效写入数据到2.0集群

背景信息

AnalyticDB for MySQLClient旨在为用户提供一个高效、简单地插入数据到AnalyticDB for MySQL 2.0集群的方法。您只需要通过接口将数据提交给AnalyticDB for MySQLClient，便可以直接插入数据到AnalyticDB for MySQL 2.0集群中。使用AnalyticDB for MySQL Client时，您无需关心分区聚合、连接池等问题，而且对数据实时写入也有更多的自主能力，不再强依赖DataHub等服务。

Maven repositories

通过Maven管理配置新的SDK版本，Maven的配置信息如下：

<dependency>
  <groupId>com.alibaba.cloud.analyticdb</groupId>
  <artifactId>adbclient</artifactId>
  <version>1.0.2</version>
</dependency>

接口列表

DatabaseConfig类

Row类

AdbClient类

ColumnInfo类

AdbClientException错误码

示例代码

public class AdbClientUsage {
        public void demo() {
            DatabaseConfig databaseConfig = new DatabaseConfig();
            // AnalyticDB MySQL 2.0集群的主机名或者URL
            databaseConfig.setHost("100.100.100.100");
            // AnalyticDB MySQL 2.0集群的连接端口号
            databaseConfig.setPort(8888);
            // AnalyticDB MySQL 2.0集群的所属账号的AccessKey ID
            databaseConfig.setUser("your db username");
            // AnalyticDB MySQL 2.0集群的所属账号的AccessKey Secret
            databaseConfig.setPassword("your db password");
            databaseConfig.setDatabase("your db name");
            // 设置需要写入的表名列表
            List<String> tables = new ArrayList<String>();
            tables.add("your table name");
            tables.add("your table name 2");
            // 一旦new Client实例之后，表配置是不可修改的
            databaseConfig.setTable(tables);

            // 设置需要写入的表字段
            List<String> columns = new ArrayList<String>();
            columns.add("column1");
            columns.add("column2");
            // 如果是所有字段，字段列表使用columns.add("*")即可
            databaseConfig.setColumns("your table name", columns);
            databaseConfig.setColumns("your table name 2", Collections.singletonList("*"));

            // 如果出现插入失败，是否跳过
            //忽略插入失败可能导致数据丢失
            databaseConfig.setIgnoreInsertError(false);
              //如果该列的值为空，则直接设置为null即可。
            databaseConfig.setEmptyAsNull(true);
            // 使用insert ignore into方式进行插入
            databaseConfig.setInsertIgnore(true);
            // 提交时，写入AnalyticDB MySQL 2.0集群出现异常时重试的3次
            databaseConfig.setRetryTimes(3);
            // 重试间隔的时间为1s，单位是ms
            databaseConfig.setRetryIntervalTime(1000);
            // 初始化AdbClient，初始化实例之后，databaseConfig的配置信息无法再进行修改
            AdbClient adbClient = new AdbClient(databaseConfig);

            // 数据需要攒批，分多次添加，再提交
            for (int i = 0; i < 10; i++) {
                // Add row(s) to buffer. One instance for one record
                Row row = new Row(columns.size());
                // Set column
                // the column index must be same as the sequence of columns
                // the column value can be any type, internally it will be formatted according to column type
                row.setColumn(0, i); // Number value
                row.setColumn(1, "string value"); // String value
                // 如果SQL长度已满，则会在addRow或者addMap的时进行一次自动提交
                // 如果提交失败，则返回AdbClientException异常，错误码为COMMIT_ERROR_DATA_LIST
                adbClient.addRow("your table name", row);
            }
            Row row = new Row();
            row.setColumn(0, 10); // Number value
            row.setColumn(1, "2018-01-01 08:00:00"); //参数2的数据类型为 Date、Timestamp或者Time value
            adbClient.addRow("your table name 2", row);
            // Update column. Row实例可复用
            row.updateColumn(0, 11);
            row.updateColumn(1, "2018-01-02 08:00:00");
            adbClient.addRow("your table name 2", row);

            // 将map加入缓存中
            Map<String, String> rowMap = new HashMap<String, String>();
            rowMap.put("column1", "124");
            rowMap.put("column2", "string value");
            // 数据需要攒批，最好多次添加之后再进行提交
            adbClient.addMap("your table name", rowMap);

            //将缓存中的数据提交到AnalyticDB MySQL 2.0集群
            //数据成功提交到AnalyticDB MySQL 2.0集群之后，缓存中的数据会被清除
            try {
                adbClient.commit();
            } catch (Exception e) {
            } finally {
                adbClient.stop();
            }
        }
    }

注意事项

• AnalyticDB for MySQLClient本身依赖druid(1.1.10)、mysql-connector-java(5.1.45)、commons-lang3(3.4)、slf4j-api(1.7.25)和slf4j-log4j12(1.7.25)，如果在使用过程中，出现版本冲突，请检查这几个包的版本并解决冲突。

• AnalyticDB for MySQLClient SDK是非线程安全的，如果多线程调用时，需要每个线程维护自己的Client对象。

  说明

  强烈不建议多线程共用SDK实例，除了线程安全问题外，还容易使Client成为写入性能的瓶颈。

• 数据必须在调用commit成功后才算成功写入AnalyticDB for MySQL。

• 针对Client抛出的异常，调用方要根据错误码的意义自行判断如何处理。如果是数据写入有问题，可以重复提交或者记录下有问题的数据后跳过。

• 很多时候写入线程并不是越多越好，因为业务程序会涉及到攒数据的场景，所以对内存的消耗比较明显，业务调用方一定要多关注应用程序的GC情况。

• 数据攒批数量不要太小，如果太小，分区聚合写意义就不大了。最好通过消息数*消息长度≈分区数*32KB大概得出攒批的消息数。

• DatabaseConfig配置在new AdbClient成功之后是无法进行修改，所有配置项必须在Client对象初始化之前完成配置。