Pyflink UDF with ARRAY as input

Posted by Barth, Torben on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Pyflink-UDF-with-ARRAY-as-input-tp40206.html

Dear List,

 

I have  a table with the following structure

 

my_table

-- Key: String

-- List_element: ARRAY<ROW<`integer_element` INT, `string_element` STRING >>

 

I want to define a udf to extract information of the “list_element”. I do not manage to access the information of the array in the udf. I try something like:

 

@udf(result_type=DataTypes.STRING())

def get_string_element(my_list):

       my_string = ‘xxx’

        for element in my_list:

            if element.integer_element  == 2:

                my_string = element. string_element

        return my_string

 

 

table_env.create_temporary_function("get_string_element", get_string_element)

# use the function in Python Table API

my_table.select("get_string_element(List_element)")

 

Unfortunately, I cannot get it work. Does anybody have an idea how the correct way to extract the information is?

 

Any comments or ideas are very welcome.

 

Thanks

Torben

Fraport AG Frankfurt Airport Services Worldwide, 60547 Frankfurt am Main, Sitz der Gesellschaft: Frankfurt am Main, Amtsgericht Frankfurt am Main: HRB 7042, Umsatzsteuer-Identifikationsnummer: DE 114150623, Vorsitzender des Aufsichtsrates: Michael Boddenberg – Hessischer Minister der Finanzen; Vorstand: Dr. Stefan Schulte (Vorsitzender), Anke Giesen, Michael Müller, Dr. Pierre Dominique Prümm, Dr. Matthias Zieschang