Converting from table to DataStream in batch mode is indeed a problem now. But I think this will be improved soon.
Best, Kurt On Fri, Apr 9, 2021 at 6:14 PM Flavio Pompermaier <pomperma...@okkam.it> wrote: > In my real CSV I have LONG columns that can contain null values. In that > case I get a parse exception (and I would like to avoid to read it as a > string). > The ',bye' is just the way you can test that in my example (add that line > to the input csv). > If I use 'csv.null-literal' = '' it seems to work but, is it a workaround > or it is the right solution? > > Another big problem I'm having with the new APIs is that if I use > TableEnvironment tableEnv = TableEnvironment.create(envSettings); > then I can't convert a table to a datastream..I need to use > StreamTableEnvironment tableEnv = > StreamTableEnvironment.create(streamEnv, envSettings); > but in that case I can't use inBatchMode.. > > On Fri, Apr 9, 2021 at 11:44 AM Kurt Young <ykt...@gmail.com> wrote: > >> `format.ignore-first-line` is unfortunately a regression compared to the >> old one. >> I've created a ticket [1] to track this but according to current design, >> it seems not easy to do. >> >> Regarding null values, I'm not sure if I understand the issue you had. >> What do you mean by >> using ',bye' to test null Long values? >> >> [1] https://issues.apache.org/jira/browse/FLINK-22178 >> >> Best, >> Kurt >> >> >> On Fri, Apr 9, 2021 at 4:46 PM Flavio Pompermaier <pomperma...@okkam.it> >> wrote: >> >>> And another thing: in my csv I added ',bye' (to test null Long values) >>> but I get a parse error..if I add 'csv.null-literal' = '' it seems to >>> work..is that the right way to solve this problem? >>> >>> On Fri, Apr 9, 2021 at 10:13 AM Flavio Pompermaier <pomperma...@okkam.it> >>> wrote: >>> >>>> Thanks Kurt, now it works. However I can't find a way to skip the CSV >>>> header..before there was "format.ignore-first-line" but now I can't find >>>> another way to skip it. >>>> I could set csv.ignore-parse-errors to true but then I can't detect >>>> other parsing errors, otherwise I need to manually transofrm the header >>>> into a comment adding the # character at the start of the line.. >>>> How can I solve that? >>>> >>>> On Fri, Apr 9, 2021 at 4:07 AM Kurt Young <ykt...@gmail.com> wrote: >>>> >>>>> My DDL is: >>>>> >>>>> CREATE TABLE csv ( >>>>> id BIGINT, >>>>> name STRING >>>>> ) WITH ( >>>>> 'connector' = 'filesystem', >>>>> 'path' = '.....', >>>>> 'format' = 'csv' >>>>> ); >>>>> >>>>> Best, >>>>> Kurt >>>>> >>>>> >>>>> On Fri, Apr 9, 2021 at 10:00 AM Kurt Young <ykt...@gmail.com> wrote: >>>>> >>>>>> Hi Flavio, >>>>>> >>>>>> We would recommend you to use new table source & sink interfaces, >>>>>> which have different >>>>>> property keys compared to the old ones, e.g. 'connector' v.s. >>>>>> 'connector.type'. >>>>>> >>>>>> You can follow the 1.12 doc [1] to define your csv table, everything >>>>>> should work just fine. >>>>>> >>>>>> *Flink SQL> set table.dml-sync=true;* >>>>>> >>>>>> *[INFO] Session property has been set.* >>>>>> >>>>>> >>>>>> *Flink SQL> select * from csv;* >>>>>> >>>>>> *+----------------------+----------------------+* >>>>>> >>>>>> *| id | name |* >>>>>> >>>>>> *+----------------------+----------------------+* >>>>>> >>>>>> *| 3 | c |* >>>>>> >>>>>> *+----------------------+----------------------+* >>>>>> >>>>>> *Received a total of 1 row* >>>>>> >>>>>> >>>>>> *Flink SQL> insert overwrite csv values(4, 'd');* >>>>>> >>>>>> *[INFO] Submitting SQL update statement to the cluster...* >>>>>> >>>>>> *[INFO] Execute statement in sync mode. Please wait for the execution >>>>>> finish...* >>>>>> >>>>>> *[INFO] Complete execution of the SQL update statement.* >>>>>> >>>>>> >>>>>> *Flink SQL> select * from csv;* >>>>>> >>>>>> *+----------------------+----------------------+* >>>>>> >>>>>> *| id | name |* >>>>>> >>>>>> *+----------------------+----------------------+* >>>>>> >>>>>> *| 4 | d |* >>>>>> >>>>>> *+----------------------+----------------------+* >>>>>> >>>>>> *Received a total of 1 row* >>>>>> >>>>>> [1] >>>>>> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/connectors/filesystem.html >>>>>> >>>>>> Best, >>>>>> Kurt >>>>>> >>>>>> >>>>>> On Fri, Apr 9, 2021 at 3:20 AM Flavio Pompermaier < >>>>>> pomperma...@okkam.it> wrote: >>>>>> >>>>>>> Hi Till, >>>>>>> since I was using the same WITH-clause both for reading and writing >>>>>>> I discovered that overwrite is actually supported in the Sinks, while in >>>>>>> the Sources an exception is thrown (I was thinking that those properties >>>>>>> were simply ignored). >>>>>>> However the quote-character is not supported in the sinks: is this a >>>>>>> bug or is it the intended behaviour?. >>>>>>> Here is a minimal example that reproduce the problem (put in the >>>>>>> /tmp/test.csv something like '1,hello' or '2,hi'). >>>>>>> >>>>>>> import org.apache.flink.table.api.EnvironmentSettings; >>>>>>> import org.apache.flink.table.api.Table; >>>>>>> import org.apache.flink.table.api.TableEnvironment; >>>>>>> >>>>>>> public class FlinkCsvTest { >>>>>>> public static void main(String[] args) throws Exception { >>>>>>> final EnvironmentSettings envSettings = >>>>>>> >>>>>>> EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build(); >>>>>>> final TableEnvironment tableEnv = >>>>>>> TableEnvironment.create(envSettings); >>>>>>> // ExecutionEnvironment env = >>>>>>> ExecutionEnvironment.getExecutionEnvironment(); >>>>>>> // BatchTableEnvironment tableEnv = >>>>>>> BatchTableEnvironment.create(env); >>>>>>> final String tableInName = "testTableIn"; >>>>>>> final String createInTableDdl = getSourceDdl(tableInName, >>>>>>> "/tmp/test.csv"); // >>>>>>> >>>>>>> final String tableOutName = "testTableOut"; >>>>>>> final String createOutTableDdl = getSinkDdl(tableOutName, >>>>>>> "/tmp/test-out.csv"); // >>>>>>> tableEnv.executeSql(createInTableDdl); >>>>>>> tableEnv.executeSql(createOutTableDdl); >>>>>>> >>>>>>> Table tableIn = tableEnv.from(tableInName); >>>>>>> Table tableOut = tableEnv.from(tableOutName); >>>>>>> tableIn.insertInto(tableOutName); >>>>>>> // tableEnv.toDataSet(table, Row.class).print(); >>>>>>> tableEnv.execute("TEST read/write"); >>>>>>> >>>>>>> } >>>>>>> >>>>>>> private static String getSourceDdl(String tableName, String >>>>>>> filePath) { >>>>>>> return "CREATE TABLE " + tableName + " (\n" + // >>>>>>> " `id` BIGINT,\n" + // >>>>>>> " `name` STRING) WITH (\n" + // >>>>>>> " 'connector.type' = 'filesystem',\n" + // >>>>>>> " 'connector.property-version' = '1',\n" + // >>>>>>> " 'connector.path' = '" + filePath + "',\n" + // >>>>>>> " 'format.type' = 'csv',\n" + // >>>>>>> " 'format.field-delimiter' = ',',\n" + // >>>>>>> // " 'format.write-mode' = 'OVERWRITE',\n" + // NOT SUPPORTED >>>>>>> " 'format.property-version' = '1',\n" + // >>>>>>> " 'format.quote-character' = '\"',\n" + // >>>>>>> " 'format.ignore-first-line' = 'false'" + // >>>>>>> ")"; >>>>>>> } >>>>>>> >>>>>>> private static String getSinkDdl(String tableName, String >>>>>>> filePath) { >>>>>>> return "CREATE TABLE " + tableName + " (\n" + // >>>>>>> " `id` BIGINT,\n" + // >>>>>>> " `name` STRING) WITH (\n" + // >>>>>>> " 'connector.type' = 'filesystem',\n" + // >>>>>>> " 'connector.property-version' = '1',\n" + // >>>>>>> " 'connector.path' = '" + filePath + "',\n" + // >>>>>>> " 'format.type' = 'csv',\n" + // >>>>>>> " 'format.field-delimiter' = ',',\n" + // >>>>>>> " 'format.num-files' = '1',\n" + // >>>>>>> " 'format.write-mode' = 'OVERWRITE',\n" + // SUPPORTED >>>>>>> (sinks only) >>>>>>> " 'format.quote-character' = '\"',\n" + // NOT SUPPORTED >>>>>>> " 'format.property-version' = '1'\n" + // >>>>>>> ")"; >>>>>>> } >>>>>>> } >>>>>>> >>>>>>> Thanks for the support, >>>>>>> Flavio >>>>>>> >>>>>>> >>>>>>> On Thu, Apr 8, 2021 at 7:05 PM Till Rohrmann <trohrm...@apache.org> >>>>>>> wrote: >>>>>>> >>>>>>>> Hi Flavio, >>>>>>>> >>>>>>>> I tried to execute the code snippet you have provided and I could >>>>>>>> not reproduce the problem. >>>>>>>> >>>>>>>> Concretely I am running this code: >>>>>>>> >>>>>>>> final EnvironmentSettings envSettings = >>>>>>>> EnvironmentSettings.newInstance() >>>>>>>> .useBlinkPlanner() >>>>>>>> .inStreamingMode() >>>>>>>> .build(); >>>>>>>> final TableEnvironment tableEnv = >>>>>>>> TableEnvironment.create(envSettings); >>>>>>>> >>>>>>>> tableEnv.fromValues("foobar").execute().await(); >>>>>>>> >>>>>>>> Am I missing something? Maybe you can share a minimal but fully >>>>>>>> working example where the problem occurs. Thanks a lot. >>>>>>>> >>>>>>>> Cheers, >>>>>>>> Till >>>>>>>> >>>>>>>> On Thu, Apr 8, 2021 at 11:25 AM Flavio Pompermaier < >>>>>>>> pomperma...@okkam.it> wrote: >>>>>>>> >>>>>>>>> Any help here? Moreover if I use the DataStream APIs there's no >>>>>>>>> left/right outer join yet..are those meant to be added in Flink 1.13 >>>>>>>>> or >>>>>>>>> 1.14? >>>>>>>>> >>>>>>>>> On Wed, Apr 7, 2021 at 12:27 PM Flavio Pompermaier < >>>>>>>>> pomperma...@okkam.it> wrote: >>>>>>>>> >>>>>>>>>> Hi to all, >>>>>>>>>> I'm testing writing to a CSV using Flink 1.13 and I get the >>>>>>>>>> following error: >>>>>>>>>> >>>>>>>>>> The matching candidates: >>>>>>>>>> org.apache.flink.table.sinks.CsvBatchTableSinkFactory >>>>>>>>>> Unsupported property keys: >>>>>>>>>> format.quote-character >>>>>>>>>> >>>>>>>>>> I create the table env using this: >>>>>>>>>> >>>>>>>>>> final EnvironmentSettings envSettings = >>>>>>>>>> EnvironmentSettings.newInstance()// >>>>>>>>>> .useBlinkPlanner()// >>>>>>>>>> // .inBatchMode()// >>>>>>>>>> .inStreamingMode()// >>>>>>>>>> .build(); >>>>>>>>>> final TableEnvironment tableEnv = >>>>>>>>>> TableEnvironment.create(envSettings); >>>>>>>>>> >>>>>>>>>> The error is the same both with inBatchMode and inStreamingMode. >>>>>>>>>> Is this really not supported or am I using the wrong API? >>>>>>>>>> >>>>>>>>>> Best, >>>>>>>>>> Flavio >>>>>>>>>> >>>>>>>>>
