@Test public void sort() { List<ETLRow> rows = new ArrayList<>(); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.INSERT, Arrays.asList(new ETLColumn("id", "1", "", "1", true), new ETLColumn("name", "zkevin", "", "zkevin", true) ), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.UPDATE, Arrays.asList(new ETLColumn("id", "1", "1", "1", true), new ETLColumn("name", "zkevin00", "zkevin", "zkevin00", true)), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.INSERT, Arrays.asList(new ETLColumn("id", "2", "", "2", true), new ETLColumn("name", "zkevin1", "", "zkevin1", true) ), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.UPDATE, Arrays.asList(new ETLColumn("id", "2", "2", "2", true), new ETLColumn("name", "zkevin11", "zkevin1", "zkevin11", true) ), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.INSERT, Arrays.asList(new ETLColumn("id", "3", "", "3", true), new ETLColumn("name", "zkevin", "", "zkevin", true) ), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.INSERT, Arrays.asList(new ETLColumn("id", "4", "", "4", true), new ETLColumn("name", "zkevin", "", "zkevin", true) ), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.INSERT, Arrays.asList(new ETLColumn("id", "5", "", "5", true), new ETLColumn("name", "zkevin", "", "zkevin", true) ), null, null)); rows.add(new ETLRow(0, 0, "s", "t", MessageAction.INSERT, Arrays.asList(new ETLColumn("id", "6", "", "6", true), new ETLColumn("name", "zkevin", "", "zkevin", true) ), null, null)); ETLBucket bucket = new ETLBucket("0", rows, null); new JdbcMultiThreadLoader().sort(bucket); Assert.assertEquals(bucket.getParallelRows().stream().flatMapToLong(r -> LongStream.of(r.size())).sum(), rows.size()); }
@Override public void sort(ETLBucket bucket) { Map<List<String>, List<ETLRow>> tables = new HashMap<>(); bucket.getRows().stream().forEach(row -> { tables.compute(Arrays.asList(row.getFinalSchema(), row.getFinalTable()), (k, v) -> { if (null == v) v = new ArrayList<>(); v.add(row); return v; }); }); tables.forEach((k, v) -> { int span = 500; for (int i = 0; i < v.size(); i += span) { int endIndex = i + span; endIndex = endIndex >= v.size() ? v.size() : endIndex; groupRows(bucket, v.subList(i, endIndex)); } }); }
JdbcMultiThreadLoader extends BaseJdbcLoader { @Override public void sort(ETLBucket bucket) { Map<List<String>, List<ETLRow>> tables = new HashMap<>(); bucket.getRows().stream().forEach(row -> { tables.compute(Arrays.asList(row.getFinalSchema(), row.getFinalTable()), (k, v) -> { if (null == v) v = new ArrayList<>(); v.add(row); return v; }); }); tables.forEach((k, v) -> { int span = 500; for (int i = 0; i < v.size(); i += span) { int endIndex = i + span; endIndex = endIndex >= v.size() ? v.size() : endIndex; groupRows(bucket, v.subList(i, endIndex)); } }); } }
JdbcMultiThreadLoader extends BaseJdbcLoader { @Override public void sort(ETLBucket bucket) { Map<List<String>, List<ETLRow>> tables = new HashMap<>(); bucket.getRows().stream().forEach(row -> { tables.compute(Arrays.asList(row.getFinalSchema(), row.getFinalTable()), (k, v) -> { if (null == v) v = new ArrayList<>(); v.add(row); return v; }); }); tables.forEach((k, v) -> { int span = 500; for (int i = 0; i < v.size(); i += span) { int endIndex = i + span; endIndex = endIndex >= v.size() ? v.size() : endIndex; groupRows(bucket, v.subList(i, endIndex)); } }); }  }
JdbcMultiThreadLoader extends BaseJdbcLoader { @Override public void sort(ETLBucket bucket) { Map<List<String>, List<ETLRow>> tables = new HashMap<>(); bucket.getRows().stream().forEach(row -> { tables.compute(Arrays.asList(row.getFinalSchema(), row.getFinalTable()), (k, v) -> { if (null == v) v = new ArrayList<>(); v.add(row); return v; }); }); tables.forEach((k, v) -> { int span = 500; for (int i = 0; i < v.size(); i += span) { int endIndex = i + span; endIndex = endIndex >= v.size() ? v.size() : endIndex; groupRows(bucket, v.subList(i, endIndex)); } }); }  @Override Pair<Boolean, List<DSubmitStatObject>> doLoad(ETLBucket bucket); @Override void sort(ETLBucket bucket); }
JdbcMultiThreadLoader extends BaseJdbcLoader { @Override public void sort(ETLBucket bucket) { Map<List<String>, List<ETLRow>> tables = new HashMap<>(); bucket.getRows().stream().forEach(row -> { tables.compute(Arrays.asList(row.getFinalSchema(), row.getFinalTable()), (k, v) -> { if (null == v) v = new ArrayList<>(); v.add(row); return v; }); }); tables.forEach((k, v) -> { int span = 500; for (int i = 0; i < v.size(); i += span) { int endIndex = i + span; endIndex = endIndex >= v.size() ? v.size() : endIndex; groupRows(bucket, v.subList(i, endIndex)); } }); }  @Override Pair<Boolean, List<DSubmitStatObject>> doLoad(ETLBucket bucket); @Override void sort(ETLBucket bucket);  }