DataStream API: Best way for reading csv file

[Jan Oelschlegel <oe...@integration-factory.de>]

Hi ,

i'm looking for an comfortable way to read a CSV file with the DataStream API in Flink 1.11.

Without using the Table/SQL-API before.

This is my first approach:

val typeInfo = TypeInformation.of(classOf[CovidEvent]).asInstanceOf[PojoTypeInfo[CovidEvent]]

val csvInputFormat = new PojoCsvInputFormat(new Path(sourcePath), "\n", ",", typeInfo)

env.readFile(csvInputFormat, sourcePath)


But then I get the following error because I want to parse a string to Date directly:

Exception in thread "main" java.lang.IllegalArgumentException: The type 'java.util.Date' is not supported for the CSV input format.


Or do I have to read it as String, then map it with some Jackson Parser and then call assignTimestampsAndWatermarks function on that?


Thanks four your help.


Best,
Jan

HINWEIS: Dies ist eine vertrauliche Nachricht und nur f?r den Adressaten bestimmt. Es ist nicht erlaubt, diese Nachricht zu kopieren oder Dritten zug?nglich zu machen. Sollten Sie diese Nachricht irrt?mlich erhalten haben, bitte ich um Ihre Mitteilung per E-Mail oder unter der oben angegebenen Telefonnummer.

Re: DataStream API: Best way for reading csv file

[Dawid Wysakowicz <dw...@apache.org>]

Hi Jan,

First of all I'd rather recommend Table API for processing structured data.

However if you are convinced you want to use the DataStream API, the
CsvInputFormat supports the java.sql.Date type. You can try that or what
I would suggest is to parse the Date field as string and then parse it
e.g. to one of the java.time classes. You could do that e.g. with the
RowCsvInputFormat or TupleCsvInputFormat.

Hope that helps.

Best,

Dawid

On 22/01/2021 18:39, Jan Oelschlegel wrote:
>
>  
>
> Hi ,
>
>  
>
> i’m looking for an comfortable way to read a CSV file with the
> DataStream API in Flink 1.11.
>
>  
>
> Without using the Table/SQL-API before.
>
>  
>
> This is my first approach:
>
>  
>
> val typeInfo =
> TypeInformation./of/(/classOf/[CovidEvent]).asInstanceOf[PojoTypeInfo[CovidEvent]]
>
> val csvInputFormat = new PojoCsvInputFormat(new Path(sourcePath),
> "\n", ",", typeInfo)
>
> env.readFile(csvInputFormat, sourcePath)
>
>  
>
>  
>
> But then I get the following error because I want to parse a string to
> Date directly:
>
>  
>
> Exception in thread "main" java.lang.IllegalArgumentException: The
> type 'java.util.Date' is not supported for the CSV input format.
>
>  
>
>  
>
> Or do I have to read it as String, then map it with some Jackson
> Parser and then call /assignTimestampsAndWatermarks /function on that?
>
>  
>
>  
>
> Thanks four your help.
>
>  
>
>  
>
> Best,
>
> Jan
>
>  
>
> HINWEIS: Dies ist eine vertrauliche Nachricht und nur für den
> Adressaten bestimmt. Es ist nicht erlaubt, diese Nachricht zu kopieren
> oder Dritten zugänglich zu machen. Sollten Sie diese Nachricht
> irrtümlich erhalten haben, bitte ich um Ihre Mitteilung per E-Mail
> oder unter der oben angegebenen Telefonnummer.