Hello I am trying to create a row using the table api with p Apache Flink #troubleshooting

Hello! I am trying to create a row using the tabl...

Kevin L

04/19/2023, 6:17 PM

Hello! I am trying to create a row using the table api with pyflink: just as a test I am doing something simple:

Copy code

t_row = t.select(row("key", "value").alias('my_row')

I expect this schema to be (when I call

t_row.get_schema()

)

Copy code

root
 |-- my_row : ROW<`key` VARCHAR NOT NULL> NOT NULL

but what I get is (see screenshot) how to I get the expected schema?

Dian Fu

04/20/2023, 5:31 AM

I’m a little confused of your questions.

getSchema

should print the actual schema of the table. Is your question that the schema printed isn’t correct or how to write a SQL statement which could your desired schema? Besides, there are three differences in the above case: •

my_row

_c0

• `ROW<`key` VARCHAR NOT NULL> NOT NULL` VS `ROW<`key` VARCHAR NOT NULL,

value

VARCHAR NOT NULL> NOT NULL` • VARCHAR VS CHAR(3) / CHAR(5) Which ones do you really care about?

Thijs van de Poll

04/20/2023, 12:07 PM

I noticed the same before. I don’t think that instantiating a row like this works. What it will do is put the column

key

at key

f0

and put the columns

value

at key

f0

. I also think this is confusing. Instead, do:

Copy code

t.select(row("col1", "col2").cast(DataTypes.ROW([DataTypes.FIELD("key_name_col1", data_type_col1), DataTypes.FIELD("key_name_cols2", data_type_col2)]).alias("my_row"))

Let me know if it helps.

👍 1

Kevin L

04/20/2023, 5:29 PM

Thank you for your help! Thijs solution worked for me!

👍 1

Kevin L

04/20/2023, 5:37 PM

My bad Dian, I quickly threw my question together, I can see how what I said wasn’t super clear. Really what I wanted to do was to override the row field names. I was expecting the

row()

expression to work similarly to the

map_()

expression (see docs). The row() docs imply that it works similarly to

map_()

based on the example they give, but it doesn’t work that way I suppose. Rather, it looks like you have to specify the ROW values via the row() expression, which assigns default field names

f0, f1, f2 …

to each value. Then using Thijs snippet, you can override those field names via casting. If anyone knows of a one-liner here that would be greatly appreciated as well

👍 1

3 Views

Open in Slack

Previous Next