analizar dinámicamente json en flink map
Estoy usando flink para analizar dinámicamente datos de tipo json, keyby y sumar con la columna dada, en mi mapFunction, convierto json a case class, pero el flujo de resultados no obtiene el compilador en keyBy function, obtuve el error Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: This type (GenericType<java.lang.Object>) cannot be used as key..my code Me gusta esto
//conf.properties
columns=a:String,b:Int,c:String,d:Long
declusteringColumns=a,c
statsColumns=b
//main function
stream.map(new MapFunc)
.keyBy(declusteringColumns(0), declusteringColumns.drop(0).toSeq: _*)
.sum(statsColumns)
class MapFunc extends RichMapFunction[String,Any]{
var clazz:Class[_]=_
override def open(parameters: Configuration): Unit = {
import scala.reflect.runtime.universe
import scala.tools.reflect.ToolBox
val tb = universe.runtimeMirror(universe.getClass.getClassLoader).mkToolBox()
clazz = tb.compile(tb.parse(
"""|case class Test(a:String,b:Int,c:String,d:Long){}
|scala.reflect.classTag[Test].runtimeClass"""
.stripMargin)).apply.asInstanceOf[Class[_]]
}
override def map(value: String) {
val tmp = JSON.parseObject(value)
val values = Utils.loadProperties("columns").split(",").map(y => {
val name = y.substring(0, y.indexOf(":"))
val tpe = y.substring(y.indexOf(":") + 1)
tpe.toLowerCase match {
case "string" => tmp.getString(name)
case "int" => tmp.getInteger(name)
case "long" => tmp.getLong(name)
case _ => null}}).toSeq
clazz.getConstructors()(0).newInstance(values: _*)
}}
¿Cómo puedo convertir json a case class o tuple?
Respuestas
En realidad, parecía que la excepción
org.apache.flink.api.common.InvalidProgramException:
This type (GenericType<Test>) cannot be used as key
permanece incluso para la clase de caso ordinario (no generado mediante reflexión)
case class Test(a: String, b: Int, c: String, d: Long)
El primer problema es que esta clase de caso no es un POJO
https://ci.apache.org/projects/flink/flink-docs-stable/dev/types_serialization.html#pojos
POJOs
Flink trata las clases Java y Scala como un tipo de datos POJO especial si cumplen los siguientes requisitos:
La clase debe ser pública.
Debe tener un constructor público sin argumentos (constructor predeterminado).
Todos los campos son públicos o deben ser accesibles a través de las funciones getter y setter. Para un campo llamado foo, los métodos getter y setter deben llamarse getFoo () y setFoo ().
El tipo de un campo debe ser compatible con un serializador registrado.
Entonces deberías reemplazar
case class Test(a: String, b: Int, c: String, d: Long)
con
import scala.beans.BeanProperty
case class Test(
@BeanProperty var a: String,
@BeanProperty var b: Int,
@BeanProperty var c: String,
@BeanProperty var d: Long) {
def this() = {
this(null, 0, null, 0)
}
}
El segundo problema puede ser que Flink no permite POJO de clase interna que no sean clases internas estáticas, pero la caja de herramientas reflectante genera una clase local anidada en un método
Reglas para tipos de POJO
Flink reconoce un tipo de datos como un tipo POJO (y permite la referencia de campo "por nombre") si se cumplen las siguientes condiciones:
- La clase es pública e independiente (sin clase interna no estática)
- La clase tiene un constructor público sin argumentos.
- Todos los campos no estáticos y no transitorios de la clase (y todas las superclases) son públicos (y no finales) o tienen un método getter y setter público que sigue las convenciones de nomenclatura de los beans de Java para getters y setters.
Aquí está la versión descompilada del código generado por la caja de herramientas
public final class __wrapper$1$a077cb72a4ee423291aac7dfb47454b9$ {
public Object wrapper() {
new LazyRef();
class Test$1 implements Product, Serializable {
private String a;
private int b;
private String c;
private long d;
...
}
return scala.reflect.package..MODULE$.classTag(scala.reflect.ClassTag..MODULE$.apply(Test$1.class)).runtimeClass();
}
...
}
El código descompilado completo:
https://gist.github.com/DmytroMitin/f1554ad833ea1bb9eb97947ae872d220
Entonces, es posible que si es realmente necesario generar una clase para Flink, se deba generar manualmente en lugar de a través de la caja de herramientas.
https://www.reddit.com/r/scala/comments/gfcmul/compile_scala_source_from_string_and/
https://www.reddit.com/r/scala/comments/jckld2/is_there_a_way_to_load_scala_code_at_runtime/
https://stackoverflow.com/questions/60783153/tensorflow-in-scala-reflection/
Pero el código con una clase generada manualmente
https://gist.github.com/DmytroMitin/e33cd244b37f9b33b67f7ac3e6609d39
todavía lanza This type (GenericType<java.lang.Object>) cannot be used as key.
Supongo que la razón es la siguiente (y este es el tercer problema ).
El código con la clase de caso ordinario (no generado) parece funcionar
https://gist.github.com/DmytroMitin/af426d4578dd5e76c9e0d344e6f079ce
Pero si reemplazamos el tipo Testcon Any, arrojaThis type (GenericType<java.lang.Object>) cannot be used as key
https://gist.github.com/DmytroMitin/a23e45a546790630e838e60c7206adcd
Y con la reflexión no podemos devolver nada más que Any.
Ahora estoy creando TypeInformation[Test]dentro de mi código generado, esto parece arreglarse This type (GenericType<java.lang.Object>) cannot be used as keypero ahora tengo
org.apache.flink.api.common.InvalidProgramException: UTF-8 is not serializable.
The object probably contains or references non serializable fields.
https://gist.github.com/DmytroMitin/16d312dbafeae54518f7ac2c490426b0
Resolví el problema InvalidProgramException: UTF-8 is not serializableanotando campos de MapFunccon@transient
https://gist.github.com/DmytroMitin/f2f859273075370c4687a30e0c3a2431
En realidad, parecía que si creamos TypeInformationcódigo interno generado, entonces la caja de herramientas es suficiente
import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import scala.reflect.runtime
import scala.reflect.runtime.universe._
import scala.tools.reflect.ToolBox
object App {
val toolbox = ToolBox(runtime.currentMirror).mkToolBox()
class MapFunc extends RichMapFunction[String, Any] {
var typeInfo: TypeInformation[_] = _
@transient var classSymbol: ClassSymbol = _
override def open(parameters: Configuration): Unit = {
val code =
"""|case class Test(
| @scala.beans.BeanProperty var a: String,
| @scala.beans.BeanProperty var b: Int,
| @scala.beans.BeanProperty var c: String,
| @scala.beans.BeanProperty var d: Long) {
| def this() = {
| this(null, 0, null, 0)
| }
|}""".stripMargin
val tree = toolbox.parse(code)
classSymbol = toolbox.define(tree.asInstanceOf[ImplDef]).asClass
typeInfo = toolbox.eval(
q"org.apache.flink.api.common.typeinfo.TypeInformation.of(classOf[${classSymbol.toType}])"
).asInstanceOf[TypeInformation[_]]
}
override def map(value: String): Any = {
val values = Seq("aaa", 1, "ccc", 2L) //hardcoded for now
createClassInstance(classSymbol, values: _*)
}
}
def main(args: Array[String]): Unit = {
val func = new MapFunc
func.open(new Configuration)
val classInstance = func.map("""{a: "aaa", b: 1, c: "ccc", d: 2}""")
println(classInstance) //Test(aaa,1,ccc,2)
val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream = env.socketTextStream("localhost", 9999)
val typeInfo = func.typeInfo.asInstanceOf[TypeInformation[Any]]
println(typeInfo)//PojoType<__wrapper$1$75434c8e32f541f7a87513a2ad2aa0ce.Test, fields = [a: String, b: Integer, c: String, d: Long]>
val res = stream.map(func)(typeInfo).keyBy("a", "c").sum("b")
println(res)//org.apache.flink.streaming.api.scala.DataStream@5927f904
}
def createClassInstance(classSymbol: ClassSymbol, args: Any*): Any = {
val runtimeMirror = toolbox.mirror
val classType = classSymbol.typeSignature
val constructorSymbol = classType.decl(termNames.CONSTRUCTOR).alternatives.head.asMethod
val classMirror = runtimeMirror.reflectClass(classSymbol)
val constructorMirror = classMirror.reflectConstructor(constructorSymbol)
constructorMirror(args: _*)
}
}