I have below code and it is working if all the values in json object/dictinoary is string, if any field in the json object/dictionary is integer it fails. I'm not sure how to fix this, I hope some one have any idea to achieve this. Code:

@udf(input_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), <http://DataTypes.INT|DataTypes.INT>()],
     result_type=DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING())
     )
def p_record(first_seen,last_seen,src,rec_count):
    record_map ={
        "first_seen": first_seen,
        "last_seen": last_seen,
        "srcid": src,
        "rec_count": rec_count
    }
    return record_map

query ="""select pdns_record(first_seen,last_seen,srcid,rec_count) from src_table"""

ftable = st_env.sql_query(query)
ftable.execute().print()

Error:

bytes_value = value.encode("utf-8")

AttributeError: 'int' object has no attribute 'encode'

at org.apache.beam.runners.fnexecution.control.FnApiControlClient$ResponseStreamObserver.onNext(FnApiControlClient.java:180)

at org.apache.beam.runners.fnexecution.control.FnApiControlClient$ResponseStreamObserver.onNext(FnApiControlClient.java:160)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onMessage(ServerCalls.java:262)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.Contexts$ContextualizedServerCallListener.onMessage(Contexts.java:76)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailableInternal(ServerCallImpl.java:318)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailable(ServerCallImpl.java:301)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1MessagesAvailable.runInContext(ServerImpl.java:834)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)

at <http://org.apache.beam.vendor.grpc.v1p43p2.io|org.apache.beam.vendor.grpc.v1p43p2.io>.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)

... 3 more

If this is python I think you are forgetting to register your function after you have defined it, like this

st_env.create_temporary_system_function("p_record",p_record)

At leasts that's what this aws repo says https://github.com/aws-samples/pyflink-getting-started/blob/main/pyflink-examples/UDF/udf.py But it doesn't works for me either. I've posted my errors some comments bellow yours. But let me know if you manage to make it work. Don't take my word on this cause I am no flink expert, just a random guy trying to use it as well

Thank you @Eugenio Gastelum for your response. I didn't paste register code here but I have it. I'm able to print the results if I take out "rec_count" field (which is integer). Rest of the fields are String and I don't have any problem with above code to process if all fields are string datatype. I am seeing this issue if I add any extra field with Integer type. My final goal is to prepare a json object/ dictionary like this {"first_seen":"2023-07-10 134314", "last_seen":"2023-07-10 134314", "rec_count":"2" } and map it my sink table with column pns_record. Then pns_record column should have exact dictionary type/ json object like this {"first_seen":"2023-07-10 134314", "last_seen":"2023-07-10 134314", "rec_count": 2 }. My sink table is Kafka topic.

create table table(                                                 
                                                pdns_record MAP(STRING,STRING)                                     
                                                `uuid` STRING,
                                                PRIMARY KEY (uuid) NOT ENFORCED
                                            ) WITH (
                                                'connector' = 'upsert-kafka',
                                                'property-version' = 'universal',
                                                'topic' = 'pns_topic',
                                                'properties.bootstrap.servers' = 'localhost:5092',
                                                'properties.group.id' = 'pns_toppic_gid',
                                                'key.format' = 'json',
                                                'key.json.ignore-parse-errors' = 'false',
                                                'value.format' = 'json',
                                                'value.json.fail-on-missing-field' = 'false'
                                            )

in this way when we sink to kafka topic the results should look like this {"pdns_record":{"first_seen":"2023-07-10 134314","last_seen":"2023-07-10 134314","rec_count": 2},"uuid":"6c975ce5-fa6c-4a1d-99f3-4222388c25b5"}

Try the following?

record_map ={
        "first_seen": str(first_seen),
        "last_seen": str(last_seen),
        "srcid": str(src),
        "rec_count": str(rec_count)
    }

Hi @Dian Fu thank for your response. Yes I tried this before and it works perfectly. But I don’t want rec_count in string, I want it in Integer.

As the result type is

DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()

and so you must make sure that the key/value of the map must both be of string type. Otherwise, it doesn’t know how to serialize the result.

Yes Sir, Perfect. I tried this too. But the problem is when I define the table like this after the above code the results are not correct.

create table table(                                                 
pdns_record STRING,                                    
`uuid` STRING,
PRIMARY KEY (uuid) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'property-version' = 'universal',
'topic' = 'pns_topic',
'properties.bootstrap.servers' = 'localhost:5092',
'properties.group.id' = 'pns_toppic_gid',
'key.format' = 'json',
'key.json.ignore-parse-errors' = 'false',
'value.format' = 'json',
'value.json.fail-on-missing-field' = 'false'
)

@Jagan Nalla have you tired the built-in sql function JSON_OBJECT ? It seems you may not need a udf, instead on your transformation to get the sink table:

query ="""select pdns_record(first_seen,last_seen,srcid,rec_count) from src_table"""

ftable = st_env.sql_query(query)
ftable.execute().print()

It seems you can do something like this

query ="""select JSON_OBJECT(KEY 'first_seen' VALUE first_seen, KEY 'last_seen' VALUE last_seen,KEY 'srcid' VALUE srcid,KEY 'rec_count' VALUE rec_count) as pd_record from src_table"""

ftable = st_env.sql_query(query)
ftable.execute().print()

Based on this article https://aiven.io/blog/preview-JSON-SQL-functions-apache-flink-1.15.0 Which takes from the flink's JSON_OBJECT documentation https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/systemfunctions/ However, on Flink's documentation they only have examples where the JSON they build is all strings, I didn't saw an example of some fields being string and some integers. Nonetheless it might just be missing that example but it can be possible. At least I searched for the JSON_OBJECT function on other RDS and they do use an example where they mix strings and values like the one the image attached:

@Eugenio Gastelum I think I tried this json_object too ( more optimized way instead of udf) But some how I stick to udf function. It's okay, the problem in both cases it generates the output in string format. for example, JSON_OBJECT function will return string data so we need to define the sink table column as string. But I need it as JSON output. I tried JSON_OBJECT function and defined my sink table column as string.

@Jagan Nalla Okay so it looks like the ultimate problem is that flink's table API does not supports the JSON that standard SQL does supports. Despite, weirdly enough, flink does support JSON functions in transformations which suggest flink does allows JSON objects in transformations. It just can support continuing the JSON object on the output stream. https://stackoverflow.com/a/73006222/9588300 But the problem seems to be how flink handles the cast from the JSON object (in the transformation) into the STRING data type defined in the sink table. As it introduces the backslashes in that operation. If you try this to let the sql engine to cast it instead of making flink to cast it? Like this

query ="""select CAST(JSON_OBJECT(KEY 'first_seen' VALUE first_seen, KEY 'last_seen' VALUE last_seen,KEY 'srcid' VALUE srcid,KEY 'rec_count' VALUE rec_count),STRING) as pd_record from src_table"""

ftable = st_env.sql_query(query)
ftable.execute().print()

If that still introduced backslashes at least we could use some string functions on it to substitute the backslashes for nothing

Yes, Nice idea, I tried your idea by typecasting like this but no use it still output backslash. When I print it, it does not show backslash.

cast(JSON_OBJECT(KEY 'rec_count' VALUE rec_count, KEY 'first_seen' VALUE first_seen, KEY 'last_seen' VALUE last_seen, KEY 'srcid' VALUE srcid) as string) pdns_record

So on your console it does not show backlash but when you sink it to kafka it does?

Yes Sir.

Okay I am reading the documentation and it says that sql has a type called ROW, that when the sync table uses

format=json

for values it casts that column defined as ROW data type into some data type on the JSON output stream. And it casts it into the JSON data type`object` type, which is essentially a JSON. So I believe if you define your column on the sink table as ROW data type, it might finally get sent as a normal json into the topic. So maybe try

create table %s (
    pdns_record ROW ('rec_count' INTEGER,'first_seen' STRING, 'last_seen' STRING,....<rest_of_fields>)
)
WITH (
.... <same kafka properties>
)

Awesome!

yeah try changing the JSON_OBJECT line for this

SELECT ROW(rec_count,first_seen,last_seen,srcid) FROM cte_flow_pdns_

the udf also returns the string format, we are very close. Need to figure it out how to convert SQL fields to ROW output. Trying to find the way to transform it into ROW

Even with this?

@udf(input_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), <http://DataTypes.INT|DataTypes.INT>()],
     result_type=DataTypes.ROW(DataTypes.FIELD('first_seen',DataTypes.STRING()), DataTypes.FIELD('last_seen',DataTypes.STRING()),DataTypes.FIELD('srcid',DataTypes.STRING()),DataTypes.FIELD('rec_count',DataTypes.STRING())
     )
def p_record(first_seen,last_seen,src,rec_count):
    record_map ={
        "first_seen": first_seen,
        "last_seen": last_seen,
        "srcid": src,
        "rec_count": rec_count
    }
    return record_map

query ="""select pdns_record(first_seen,last_seen,srcid,rec_count) from src_table"""

ftable = st_env.sql_query(query)
ftable.execute().print()

Tried this seeing and got different error:

@udf(input_types=[DataTypes.STRING(), DataTypes.STRING(), DataTypes.STRING(), <http://DataTypes.INT|DataTypes.INT>()],
     result_type=DataTypes.ROW([DataTypes.FIELD('first_seen',DataTypes.STRING()),DataTypes.FIELD('last_seen',DataTypes.STRING()),DataTypes.FIELD('srcid',DataTypes.STRING()),DataTypes.FIELD('rec_count',DataTypes.STRING())])
     )
def pdns_record(first_seen, last_seen, srcid, rec_count):
    record_map = {
        "first_seen": first_seen,
        "last_seen": last_seen,
        "srcid": srcid,
        "rec_count": rec_count
    }
    return record_map

Error:

AttributeError: 'dict' object has no attribute 'get_fields_by_names'

        at org.apache.beam.runners.fnexecution.control.FnApiControlClient$ResponseStreamObserver.onNext(FnApiControlClient.java:180)
        at org.apache.beam.runners.fnexecution.control.FnApiControlClient$ResponseStreamObserver.onNext(FnApiControlClient.java:160)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onMessage(ServerCalls.java:262)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.Contexts$ContextualizedServerCallListener.onMessage(Contexts.java:76)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailableInternal(ServerCallImpl.java:318)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailable(ServerCallImpl.java:301)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1MessagesAvailable.runInContext(ServerImpl.java:834)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
        at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
        ... 3 more

Okay so it seems the UDF will have to return somehow a ROW object instead of a dict. But before going that way, instead of using the udf, did replacing the

JSON_OBJECT

query for this worked?

SELECT ROW(rec_count,first_seen,last_seen,srcid) FROM cte_flow_pdns

This worked perfectly... Amazing

Congrats! @Jagan Nalla