You should be able to use the "print" sink. Remember though that the "print" sink prints into the stdout/stderr of TaskManagers, not the Client, where you submit the query. This is different from the TableResult, which collects results in the client. BTW, for printing you can use TableResult#print, which will nicely format your results.
Best,
Dawid
On 29/10/2020 16:13, Ruben Laguna wrote:
> How can I use the Table [Print SQL connector][1]? I tried the
> following (batch mode) but it does not give any output:
>
>
> EnvironmentSettings settings =
> EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
> TableEnvironment tEnv = TableEnvironment.create(settings);
>
> final LocalDateTime DATE_TIME = LocalDateTime.of(2020, 1, 1, 0, 0);
>
> Table transactions =
> tEnv.fromValues(
> DataTypes.ROW(
> DataTypes.FIELD("account_id", DataTypes.BIGINT()),
> DataTypes.FIELD("amount", DataTypes.BIGINT()),
> DataTypes.FIELD("transaction_time",
> DataTypes.TIMESTAMP(3))),
> Row.of(1, 188, DATE_TIME.plusMinutes(12)),
> Row.of(2, 374, DATE_TIME.plusMinutes(47)),
> Row.of(3, 112, DATE_TIME.plusMinutes(36)),
> Row.of(4, 478, DATE_TIME.plusMinutes(3)),
> Row.of(5, 208, DATE_TIME.plusMinutes(8)),
> Row.of(1, 379, DATE_TIME.plusMinutes(53)),
> Row.of(2, 351, DATE_TIME.plusMinutes(32)),
> Row.of(3, 320, DATE_TIME.plusMinutes(31)),
> Row.of(4, 259, DATE_TIME.plusMinutes(19)),
> Row.of(5, 273, DATE_TIME.plusMinutes(42)));
> tEnv.executeSql("CREATE TABLE print_table(account_id BIGINT, amount
> BIGINT, transaction_time TIMESTAMP) WITH ('connector' = 'print')");
>
> transactions.executeInsert("print_table");
>
>
> I can "materialize" the result manually and print them out with :
>
> for (Row row : materialize(transactions.execute())) {
> System.out.println(row);
> }
>
> private static List<Row> materialize(TableResult results) {
> try (CloseableIterator<Row> resultIterator = results.collect()) {
> return StreamSupport
>
> .stream(Spliterators.spliteratorUnknownSize(resultIterator,
> Spliterator.ORDERED), false)
> .collect(Collectors.toList());
> } catch (Exception e) {
> throw new RuntimeException("Failed to materialize results", e);
> }
> }
>
>
> But I would like to know why I can't just use the Print sink.
>
> I've tried with `.inBatchMode()` and with `inStreamingMode()`, so I
> don't thinks it's that.
>
> Does anybody know of any working example involving the print connector?
>
>
>
> [1]:
> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/print.html
signature.asc
Description: OpenPGP digital signature
