pyflink-scala udf-如何在表api中转换scalaMap？

abithluo 于 2021-06-21 发布在 Flink

关注(0)|答案(1)|浏览(565)

我在画Map Map[String,String] scala自定义项的对象输出( scala.collection.immutable.map )到表api中的某个有效数据类型，即通过java类型( java.util.Map )这里推荐：flink表api&sql和Map类型（scala）。然而我得到下面的错误。
你知道正确的方法吗？如果是，是否有方法将转换泛化为类型为的（嵌套）scala对象 Map[String,Any] ?
代码
scala自定义项

class dummyMap() extends ScalarFunction {
  def eval() = {
    val whatevermap = Map("key1" -> "val1", "key2" -> "val2")
    whatevermap.asInstanceOf[java.util.Map[java.lang.String,java.lang.String]]
  }
}

下沉

my_sink_ddl = f"""
    create table mySink (
        output_of_dummyMap_udf MAP<STRING,STRING>
    ) with (
        ...
    )
"""

错误

Py4JJavaError: An error occurred while calling o430.execute.
: org.apache.flink.table.api.ValidationException: Field types of query result and registered TableSink `default_catalog`.`default_database`.`mySink` do not match.
Query result schema: [output_of_my_scala_udf: GenericType<java.util.Map>]
TableSink schema:    [output_of_my_scala_udf: Map<String, String>]

谢谢！

apache-flink pyflink

来源：https://stackoverflow.com/questions/64721849/pyflink-scala-udf-how-to-convert-scala-map-in-table-api

1条答案

按热度按时间

dhxwm5r41#

魏忠的原始答案。我只是个记者。谢谢小薇！
此时（flink 1.11），有两种方法在起作用：
当前：自定义项定义中的datatypehint+自定义项注册的sql
过时：重写udf定义中的getresulttype+t琰env.register琰java琰函数以进行udf注册
代码
scala自定义项

package com.dummy

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.annotation.DataTypeHint
import org.apache.flink.table.api.Types
import org.apache.flink.table.functions.ScalarFunction
import org.apache.flink.types.Row

class dummyMap extends ScalarFunction {

  // If the udf would be registered by the SQL statement, you need add this typehint
  @DataTypeHint("ROW<s STRING,t STRING>")
  def eval(): Row = {

    Row.of(java.lang.String.valueOf("foo"), java.lang.String.valueOf("bar"))

  }

  // If the udf would be registered by the method 'register_java_function', you need override this
  // method.
  override def getResultType(signature: Array[Class[_]]): TypeInformation[_] = {
    // The type of the return values should be TypeInformation
    Types.ROW(Array("s", "t"), Array[TypeInformation[_]](Types.STRING(), Types.STRING()))
  }
}

python代码

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment

s_env = StreamExecutionEnvironment.get_execution_environment()
st_env = StreamTableEnvironment.create(s_env)

# load the scala udf jar file, the path should be modified to yours

# or your can also load the jar file via other approaches

st_env.get_config().get_configuration().set_string("pipeline.jars", "file:///Users/zhongwei/the-dummy-udf.jar")

# register the udf via

st_env.execute_sql("CREATE FUNCTION dummyMap AS 'com.dummy.dummyMap' LANGUAGE SCALA")

# or register via the method

# st_env.register_java_function("dummyMap", "com.dummy.dummyMap")

# prepare source and sink

t = st_env.from_elements([(1, 'hi', 'hello'), (2, 'hi', 'hello')], ['a', 'b', 'c'])
st_env.execute_sql("""create table mySink (
        output_of_my_scala_udf ROW<s STRING,t STRING>
    ) with (
        'connector' = 'print'
    )""")

# execute query

t.select("dummyMap()").execute_insert("mySink").get_job_client().get_job_execution_result().result()

赞(0）回复(0）举报 2021-06-21

我来回答

pyflink-scala udf-如何在表api中转换scalaMap？

1条答案

相关问题

热门标签

最新问答