Hi Steve,
Unfortunately the information you posted still does not explain how you
ended up with *RAW('java.util.Map', ?)* for your input type. Would be
best if you could share an example that I could use to reproduce it.
I tried putting down some potential approaches:
I tested it with a class generated from an avsc:
{"namespace": "com.ververica.avro.generated",
"type": "record",
"name": "Address",
"fields": [
{"name": "num", "type": "int"},
{"name": "street", "type": {
"type": "map",
"values" : "string",
"default": {}
}}
]
}
which has two fields:
@Deprecated public int num;
@Deprecated public java.util.Map<java.lang.String,java.lang.String>
street;
1) From the description you posted the UrlParameters (street in my case)
field should have *LEGACY('RAW', 'ANY<java.util.Map>')* type.
root
|-- num: INT
|-- street: LEGACY('RAW', 'ANY<java.util.Map>')
2) Using the new type system
A more seamless integration of the DataStream <> Table integration is
still under development. You can check FLIP-136[1] for it. Therefore
you'd need to adjust your types in the input DataStream. Bare in mind
this approach changes the way the type is serialized from an Avro based
to custom Flink's POJO serialization.
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
Map<String, TypeInformation<?>> fieldTypes = new HashMap<>();
fieldTypes.put("num", BasicTypeInfo.INT_TYPE_INFO);
fieldTypes.put("street", Types.MAP(Types.STRING, Types.STRING));
SingleOutputStreamOperator<Address> elements = env.fromElements(
Address.newBuilder()
.setNum(1)
.setStreet(new HashMap<>())
.build()
)
.returns(
Types.POJO(
Address.class,
fieldTypes
)
);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(
env,
EnvironmentSettings.newInstance().useBlinkPlanner().build());
tEnv.createTemporaryView("test", elements);
tEnv.from("test").select(call(Func.class,
$("street"))).execute().print();
}
public static class Func extends ScalarFunction {
@FunctionHint(
input = {@DataTypeHint(value = "MAP<STRING, STRING>")},
output = @DataTypeHint("STRING")
)
public String eval(final Map<String, String> map) {
// business logic
return "ABC";
}
}
3) Using the legacy types approach you can query that field like this:
public static class LegacyFunc extends ScalarFunction {
public String eval(final Map<String, String> map) {
// business logic
return "ABC";
}
}
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
SingleOutputStreamOperator<Address> elements = env.fromElements(
Address.newBuilder()
.setNum(1)
.setStreet(new HashMap<>())
.build()
);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(
env,
EnvironmentSettings.newInstance().useBlinkPlanner().build());
tEnv.createTemporaryView("test", elements);
tEnv.registerFunction("legacyFunc", new LegacyFunc());
tEnv.from("test").select(call("legacyFunc",
$("street"))).execute().print();
Best,
Dawid
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-136%3A++Improve+interoperability+between+DataStream+and+Table+API
On 09/11/2020 02:30, Steve Whelan wrote:
Hi Dawid,
Just wanted to bump this thread in case you had any thoughts.
Thanks,
Steve
On Thu, Oct 29, 2020 at 2:42 PM Steve Whelan <swhe...@jwplayer.com
<mailto:swhe...@jwplayer.com>> wrote:
For some background, I am upgrading from Flink v1.9 to v1.11. So
what I am about to describe is our implementation on v1.9, which
worked. I am trying to achieve the same functionality on v1.11.
I have a DataStream whose type is an avro generated POJO, which
contains a field *UrlParameters* that is of type *Map<String,
String>*. I register this stream as a view so I can perform SQL
queries on it. One of the queries contains the UDF I have
previously posted. It appears that in the conversion to a view,
the type of *UrlParameters* is being converted into
*RAW('java.util.Map', ?)*.
*Code on v1.9*
DataStream pings = // a Kafka stream source deserialized into an
avro generated POJO
tableEnvironment.registerDataStream("myTable", pings);
table = tableEnvironment.sqlQuery("SELECT MAP_VALUE(UrlParameters,
'some_key') FROM myTable");
// tablesinks...
/The produced type of my deserializer is:/
@Override
public TypeInformation<Ping> getProducedType() {
// Ping.class is an avro generated POJO
return TypeInformation.of(Ping.class);
}
/Scalar UDF MAP_VALUE:/
public static String eval(final Map<String, String> map, final
String key) {
return map.get(key);
}
I an using a UDF to access fields in the *UrlParameters* map
because if I try to access them directly in the SQL (i.e.
`*UrlParameters['some_key']*`), I get the below exception. This
stackoverflow[1] had suggested the UDF as a work around.
Caused by: org.apache.flink.table.api.TableException: Type is not
supported: ANY
at
org.apache.flink.table.planner.calcite.FlinkTypeFactory$.toLogicalType(FlinkTypeFactory.scala:551)
at
org.apache.flink.table.planner.codegen.ExprCodeGenerator.visitCall(ExprCodeGenerator.scala:478)
at
org.apache.flink.table.planner.codegen.ExprCodeGenerator.visitCall(ExprCodeGenerator.scala:53)
at org.apache.calcite.rex.RexCall.accept(RexCall.java:288)
at
org.apache.flink.table.planner.codegen.ExprCodeGenerator.$anonfun$visitCall$1(ExprCodeGenerator.scala:490)
This above implementation worked successfully on v1.9. We use a
stream source instead of a table source b/c we do other non-SQL
type things with the stream.
*Code on v1.11*
The following is the implementation on v1.11 which does not work.
I was using the Old Planner on v1.9 but have switched to the Blink
Planner on v1.11, in case that has any relevance here.
DataStream pings = // a Kafka stream source deserialized into an
avro generated POJO object
tableEnvironment.createTemporaryView("myTable", pings);
table = tableEnvironment.sqlQuery("SELECT MAP_VALUE(UrlParameters,
'some_key') FROM myTable");
// tablesinks...
The UDF referenced above produced the below error. So I assumed
adding DataTypeHints was the way to solve it but I was unable to
get that to work. That is what prompted the initial email to the ML.
Caused by: org.apache.flink.table.api.ValidationException: Invalid
input arguments. Expected signatures are:
MAP_VALUE(map => MAP<STRING, STRING>, key => STRING)
at
org.apache.flink.table.types.inference.TypeInferenceUtil.createInvalidInputException(TypeInferenceUtil.java:190)
at
org.apache.flink.table.planner.functions.inference.TypeInferenceOperandChecker.checkOperandTypesOrError(TypeInferenceOperandChecker.java:131)
at
org.apache.flink.table.planner.functions.inference.TypeInferenceOperandChecker.checkOperandTypes(TypeInferenceOperandChecker.java:89)
... 50 more
Caused by: org.apache.flink.table.api.ValidationException: Invalid
argument type at position 0. Data type MAP<STRING, STRING>
expected but RAW('java.util.Map', ?) passed.
at
org.apache.flink.table.types.inference.TypeInferenceUtil.adaptArguments(TypeInferenceUtil.java:137)
at
org.apache.flink.table.types.inference.TypeInferenceUtil.adaptArguments(TypeInferenceUtil.java:102)
at
org.apache.flink.table.planner.functions.inference.TypeInferenceOperandChecker.checkOperandTypesOrError(TypeInferenceOperandChecker.java:126)
... 51 more
I can try creating a concrete reproducible example if this
explanation isn't enough though its quite a bit with the avro POJO
and custom deserializer.
Thanks,
Steve
[1]
https://stackoverflow.com/questions/45621542/does-flink-sql-support-java-map-types
<https://stackoverflow.com/questions/45621542/does-flink-sql-support-java-map-types>
