Using RowTypeInfo with Table API

[Jens Grassel]

Hi,

I tried to create a table from a DataStream[Row] but got a (somehow
expected) error:

<---snip--->
An input of GenericTypeInfo<Row> cannot be converted to Table. Please
specify the type of the input with a RowTypeInfo.
<---snip--->

Code looks like this:

val ds: DataStream[Row] = ...
val dT = streamTableEnvironment.fromDataStream(ds, 'f1, 'f2, ...)

However I have not found an example or hints at how to actually use the
mentioned RowTypeInfo to specify a column mapping for the table api.

Can anyone shed some light on this?

Regards,

Jens

--
CTO, Wegtam GmbH, 27. Hornung 2018, 14:28
Homepage : https://www.wegtam.com

So you think that money is the root of all evil.
Have you ever asked what is the root of money?
                -- Ayn Rand

[Timo Walther]

Hi Jens,

usually the Flink extracts the type information from the generic
signature of a function. E.g. it knows the fields of a Tuple2<Integer,
String>. The row type cannot be analyzed and therefore always needs
explicit information.

You can create it with org.apache.flink.table.api.Types.ROW(...). You
can check the type of your stream using ds.getType(). You can pass
information explicitly in Scala e.g. ds.map()(Types.ROW(...))

Hope that helps.

Regards,
Timo


Am 2/27/18 um 2:34 PM schrieb Jens Grassel:

> Hi,
>
> I tried to create a table from a DataStream[Row] but got a (somehow
> expected) error:
>
> <---snip--->
> An input of GenericTypeInfo<Row> cannot be converted to Table. Please
> specify the type of the input with a RowTypeInfo.
> <---snip--->
>
> Code looks like this:
>
> val ds: DataStream[Row] = ...
> val dT = streamTableEnvironment.fromDataStream(ds, 'f1, 'f2, ...)
>
> However I have not found an example or hints at how to actually use the
> mentioned RowTypeInfo to specify a column mapping for the table api.
>
> Can anyone shed some light on this?
>
> Regards,
>
> Jens
>

[Jens Grassel]

Hi,

On Tue, 27 Feb 2018 14:43:06 +0100
Timo Walther <[hidden email]> wrote:

TW> You can create it with org.apache.flink.table.api.Types.ROW(...).
TW> You can check the type of your stream using ds.getType(). You can
TW> pass information explicitly in Scala e.g. ds.map()(Types.ROW(...))

thanks that works but now we're nailed to the types supported by
`Types` which seem to be not that many (String, Int, Decimal, etc.).

Is there a way to include custom types or at least something like UUID?

Regards,

Jens

--
CTO, Wegtam GmbH, 27. Hornung 2018, 15:07
Homepage : https://www.wegtam.com

We have lingered long enough on the shores of the Cosmic Ocean.
                -- Carl Sagan

[Timo Walther]

Hi Jens,

you can always use TypeInformation.of() for all supported Flink types.
In [1] you can also find a list of all types.

Regards,
Timo

[1]
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/Types.java


Am 2/27/18 um 3:13 PM schrieb Jens Grassel:

> Hi,
>
> On Tue, 27 Feb 2018 14:43:06 +0100
> Timo Walther <[hidden email]> wrote:
>
> TW> You can create it with org.apache.flink.table.api.Types.ROW(...).
> TW> You can check the type of your stream using ds.getType(). You can
> TW> pass information explicitly in Scala e.g. ds.map()(Types.ROW(...))
>
> thanks that works but now we're nailed to the types supported by
> `Types` which seem to be not that many (String, Int, Decimal, etc.).
>
> Is there a way to include custom types or at least something like UUID?
>
> Regards,
>
> Jens
>

[Jens Grassel]

Hi,

On Tue, 27 Feb 2018 15:20:00 +0100
Timo Walther <[hidden email]> wrote:

TW> you can always use TypeInformation.of() for all supported Flink
TW> types. In [1] you can also find a list of all types.

thank you very much for you help.

Regards,

Jens

--
CTO, Wegtam GmbH, 27. Hornung 2018, 15:23
Homepage : https://www.wegtam.com

The first myth of management is that it exists.  The second myth of
management is that success equals skill.
                -- Robert Heller

[Fabian Hueske-2]

Hi Jens,

you're not bound to the types supported by Types.
Types.ROW() accepts regular TypeInformation objects.
For example org.apache.flink.api.common.typeinfo.Types offers a few more types than org.apache.flink.table.api.Types.

Best, Fabian

2018-02-27 15:13 GMT+01:00 Jens Grassel <[hidden email]>:

Hi,

On Tue, 27 Feb 2018 14:43:06 +0100
Timo Walther <[hidden email]> wrote:

TW> You can create it with org.apache.flink.table.api.Types.ROW(...).
TW> You can check the type of your stream using ds.getType(). You can
TW> pass information explicitly in Scala e.g. ds.map()(Types.ROW(...))

thanks that works but now we're nailed to the types supported by
`Types` which seem to be not that many (String, Int, Decimal, etc.).

Is there a way to include custom types or at least something like UUID?

Regards,

Jens

--
CTO, Wegtam GmbH, 27. Hornung 2018, 15:07
Homepage : https://www.wegtam.com

We have lingered long enough on the shores of the Cosmic Ocean.
                -- Carl Sagan