Apache Flink：将DataStream写入Postgres表

Question

我试图编写一个流作业，它将数据流下沉到postgres表中。为了提供完整的信息，我的工作基于以下文章:https://tech.signavio.com/2017/postgres-flink-sink，这些文章建议使用JDBCoutputFormat。

98     ... 
99     String strQuery = "INSERT INTO public.alarm (entity, duration, first, type, windowsize) VALUES (?, ?, ?, 'dur', 6)";
100
101     JDBCOutputFormat jdbcOutput = JDBCOutputFormat.buildJDBCOutputFormat()
102      .setDrivername("org.postgresql.Driver")
103      .setDBUrl("jdbc:postgresql://localhost:5432/postgres?user=michel&password=polnareff")
104      .setQuery(strQuery)
105      .setSqlTypes(new int[] { Types.VARCHAR, Types.INTEGER, Types.VARCHAR}) //set the types
106      .finish();
107
108     DataStream<Row> rows = FilterStream
109                 .map((tuple)-> {
110                    Row row = new Row(3);                  // our prepared statement has 3 parameters
111                    row.setField(0, tuple.f0);             // first parameter is case ID
112                    row.setField(1, tuple.f1);             // second paramater is tracehash
113                    row.setField(2, f.format(tuple.f2));   // third paramater is tracehash
114                    return row;
115                 });
116
117     rows.writeUsingOutputFormat(jdbcOutput);
118
119     env.execute();
120
121     }
122 }

所以我的问题是：我错过了什么吗？我应该将插入的行提交到某个地方吗？

向你致意，伊格内修斯

Answer 1

正如Chesnay在他的评论中所说，你必须调整批处理间隔。

然而，这还不是故事的全部。如果您想获得至少一次的结果，您必须将批处理写与Flink的检查点同步。基本上，您必须将JDBCoutputFormat包装在SinkFunction中，该函数也实现了CheckPointedFunction接口。当调用snapshotstate()时，您已经将批处理写入数据库。您可以查看这个请求，它将在下一个版本中提供此功能。