Hazelcast jet 数据加载

Question

我正在尝试将数据库数据加载到 hazelcast 喷气式地图实例中

public void sourceStage(JetInstance jet, Pipeline pipeLine){
SimpleDriverDataSource dataSource = new SimpleDriverDataSource();
    dataSource.setDriver(new org.postgresql.Driver());
    dataSource.setUrl("jdbc:postgresql://localhost/development");
    dataSource.setUsername("postgres");
    dataSource.setPassword("root");
    JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
    String sql1 = "SELECT  id1 , id2 FROM public.tbl_data where name='india'";
    jdbcTemplate.query(sql1, new RowMapper<Object>() {
        @Override
        public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
            jet.getMap("t1").put(rs.getString("id1"), rs.getString("id2"));
            return null;
        }
    });
}

并再次尝试在管道中使用相同的地图作为源

        pipeLine.readFrom(Sources.map("t1")).map(map -> {
            System.err.println(map.getKey() + "---" + map.getValue());
            return map;
        }).writeTo(Sinks.logger());

我发现我无法将数据作为源读取。 但如果检查jet.getMap("t1").size() 我正在获取大小值。

Answer 1

要加载数据库数据，请使用带有JDBC source 和IMap Sink 的管道。将两者混合到一个阶段会破坏 Jet 引擎的并行执行模型。