diff --git a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java index d1d19d3f..fbb188da 100644 --- a/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java +++ b/client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbEtlService.java @@ -13,12 +13,12 @@ import java.util.concurrent.atomic.AtomicLong; import javax.sql.DataSource; -import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.rdb.config.MappingConfig.DbMapping; +import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil; import com.alibaba.otter.canal.client.adapter.support.EtlResult; import com.alibaba.otter.canal.client.adapter.support.Util; import com.google.common.base.Joiner; @@ -36,7 +36,8 @@ public class RdbEtlService { /** * 导入数据 */ - public static EtlResult importData(DataSource srcDS, DataSource targetDS, MappingConfig config, List params) { + public static EtlResult importData(DataSource srcDS, DataSource targetDS, MappingConfig config, + List params) { EtlResult etlResult = new EtlResult(); AtomicLong successCount = new AtomicLong(); List errMsg = new ArrayList<>(); @@ -53,8 +54,8 @@ public class RdbEtlService { long start = System.currentTimeMillis(); // 拼接sql - StringBuilder sql = new StringBuilder("SELECT * FROM " + dbMapping.getDatabase() + "." - + dbMapping.getTable()); + StringBuilder sql = new StringBuilder( + "SELECT * FROM " + dbMapping.getDatabase() + "." + dbMapping.getTable()); // 拼接条件 appendCondition(params, dbMapping, srcDS, sql); @@ -91,12 +92,8 @@ public class RdbEtlService { } else { sqlFinal = sql + " LIMIT " + offset + "," + cnt; } - Future future = executor.submit(() -> executeSqlImport(srcDS, - targetDS, - sqlFinal, - dbMapping, - successCount, - errMsg)); + Future future = executor + .submit(() -> executeSqlImport(srcDS, targetDS, sqlFinal, dbMapping, successCount, errMsg)); futures.add(future); } @@ -109,11 +106,11 @@ public class RdbEtlService { executeSqlImport(srcDS, targetDS, sql.toString(), dbMapping, successCount, errMsg); } - logger.info(dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 - + "s!"); + logger.info( + dbMapping.getTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!"); - etlResult.setResultMessage("导入目标表 " + SyncUtil.getDbTableName(dbMapping) + " 数据：" + successCount.get() - + " 条"); + etlResult + .setResultMessage("导入目标表 " + SyncUtil.getDbTableName(dbMapping) + " 数据：" + successCount.get() + " 条"); } catch (Exception e) { logger.error(e.getMessage(), e); errMsg.add(hbaseTable + " etl failed! ==>" + e.getMessage()); @@ -127,8 +124,8 @@ public class RdbEtlService { return etlResult; } - private static void appendCondition(List params, DbMapping dbMapping, DataSource ds, StringBuilder sql) - throws SQLException { + private static void appendCondition(List params, DbMapping dbMapping, DataSource ds, + StringBuilder sql) throws SQLException { if (params != null && params.size() == 1 && dbMapping.getEtlCondition() == null) { AtomicBoolean stExists = new AtomicBoolean(false); // 验证是否有SYS_TIME字段 @@ -145,9 +142,9 @@ public class RdbEtlService { } } catch (Exception e) { // ignore - } - return null; - } ); + } + return null; + }); if (stExists.get()) { sql.append(" WHERE SYS_TIME >= '").append(params.get(0)).append("' "); } @@ -168,13 +165,12 @@ public class RdbEtlService { private static boolean executeSqlImport(DataSource srcDS, DataSource targetDS, String sql, DbMapping dbMapping, AtomicLong successCount, List errMsg) { try { - Util.sqlRS(srcDS, sql, rs -> { - int idx = 1; + Map columnsMap = new LinkedHashMap<>(); + Map columnType = new LinkedHashMap<>(); + Util.sqlRS(targetDS, "SELECT * FROM " + SyncUtil.getDbTableName(dbMapping) + " LIMIT 1 ", rs -> { try { - boolean completed = false; - Map columnType = new LinkedHashMap<>(); ResultSetMetaData rsd = rs.getMetaData(); int columnCount = rsd.getColumnCount(); List columns = new ArrayList<>(); @@ -183,91 +179,105 @@ public class RdbEtlService { columns.add(rsd.getColumnName(i)); } - Map columnsMap = SyncUtil.getColumnsMap(dbMapping, columns); + columnsMap.putAll(SyncUtil.getColumnsMap(dbMapping, columns)); + return true; + } catch (Exception e) { + logger.error(e.getMessage(), e); + return false; + } + }); + + Util.sqlRS(srcDS, sql, rs -> { + int idx = 1; + + try { + boolean completed = false; + // if (dbMapping.isMapAll()) { // columnsMap = dbMapping.getAllColumns(); // } else { // columnsMap = dbMapping.getTargetColumns(); // } - StringBuilder insertSql = new StringBuilder(); - insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" ("); - columnsMap.forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); + StringBuilder insertSql = new StringBuilder(); + insertSql.append("INSERT INTO ").append(SyncUtil.getDbTableName(dbMapping)).append(" ("); + columnsMap + .forEach((targetColumnName, srcColumnName) -> insertSql.append(targetColumnName).append(",")); - int len = insertSql.length(); - insertSql.delete(len - 1, len).append(") VALUES ("); - int mapLen = columnsMap.size(); - for (int i = 0; i < mapLen; i++) { - insertSql.append("?,"); - } - len = insertSql.length(); - insertSql.delete(len - 1, len).append(")"); - try (Connection connTarget = targetDS.getConnection(); - PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) { - connTarget.setAutoCommit(false); + int len = insertSql.length(); + insertSql.delete(len - 1, len).append(") VALUES ("); + int mapLen = columnsMap.size(); + for (int i = 0; i < mapLen; i++) { + insertSql.append("?,"); + } + len = insertSql.length(); + insertSql.delete(len - 1, len).append(")"); + try (Connection connTarget = targetDS.getConnection(); + PreparedStatement pstmt = connTarget.prepareStatement(insertSql.toString())) { + connTarget.setAutoCommit(false); - while (rs.next()) { - pstmt.clearParameters(); + while (rs.next()) { + pstmt.clearParameters(); - // 删除数据 - Map values = new LinkedHashMap<>(); - StringBuilder deleteSql = new StringBuilder("DELETE FROM " + SyncUtil.getDbTableName(dbMapping) - + " WHERE "); - appendCondition(dbMapping, deleteSql, values, rs); - try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { - int k = 1; - for (Object val : values.values()) { - pstmt2.setObject(k++, val); - } - pstmt2.execute(); - } - - int i = 1; - for (Map.Entry entry : columnsMap.entrySet()) { - String targetClolumnName = entry.getKey(); - String srcColumnName = entry.getValue(); - if (srcColumnName == null) { - srcColumnName = targetClolumnName; + // 删除数据 + Map values = new LinkedHashMap<>(); + StringBuilder deleteSql = new StringBuilder( + "DELETE FROM " + SyncUtil.getDbTableName(dbMapping) + " WHERE "); + appendCondition(dbMapping, deleteSql, values, rs); + try (PreparedStatement pstmt2 = connTarget.prepareStatement(deleteSql.toString())) { + int k = 1; + for (Object val : values.values()) { + pstmt2.setObject(k++, val); + } + pstmt2.execute(); } - Integer type = columnType.get(targetClolumnName.toLowerCase()); + int i = 1; + for (Map.Entry entry : columnsMap.entrySet()) { + String targetClolumnName = entry.getKey(); + String srcColumnName = entry.getValue(); + if (srcColumnName == null) { + srcColumnName = targetClolumnName; + } - Object value = rs.getObject(srcColumnName); - if (value != null) { - SyncUtil.setPStmt(type, pstmt, value, i); - } else { - pstmt.setNull(i, type); + Integer type = columnType.get(targetClolumnName.toLowerCase()); + + Object value = rs.getObject(srcColumnName); + if (value != null) { + SyncUtil.setPStmt(type, pstmt, value, i); + } else { + pstmt.setNull(i, type); + } + + i++; } - i++; - } + pstmt.execute(); + if (logger.isTraceEnabled()) { + logger.trace("Insert into target table, sql: {}", insertSql); + } - pstmt.execute(); - if (logger.isTraceEnabled()) { - logger.trace("Insert into target table, sql: {}", insertSql); + if (idx % dbMapping.getCommitBatch() == 0) { + connTarget.commit(); + completed = true; + } + idx++; + successCount.incrementAndGet(); + if (logger.isDebugEnabled()) { + logger.debug("successful import count:" + successCount.get()); + } } - - if (idx % dbMapping.getCommitBatch() == 0) { + if (!completed) { connTarget.commit(); - completed = true; - } - idx++; - successCount.incrementAndGet(); - if (logger.isDebugEnabled()) { - logger.debug("successful import count:" + successCount.get()); } } - if (!completed) { - connTarget.commit(); - } - } - } catch (Exception e) { - logger.error(dbMapping.getTable() + " etl failed! ==>" + e.getMessage(), e); - errMsg.add(dbMapping.getTable() + " etl failed! ==>" + e.getMessage()); - } - return idx; - } ); + } catch (Exception e) { + logger.error(dbMapping.getTable() + " etl failed! ==>" + e.getMessage(), e); + errMsg.add(dbMapping.getTable() + " etl failed! ==>" + e.getMessage()); + } + return idx; + }); return true; } catch (Exception e) { logger.error(e.getMessage(), e); @@ -278,8 +288,8 @@ public class RdbEtlService { /** * 拼接目标表主键where条件 */ - private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map values, ResultSet rs) - throws SQLException { + private static void appendCondition(DbMapping dbMapping, StringBuilder sql, Map values, + ResultSet rs) throws SQLException { // 拼接主键 for (Map.Entry entry : dbMapping.getTargetPk().entrySet()) { String targetColumnName = entry.getKey();