analisar dinamicamente json no mapa flink
Estou usando o flink para analisar dinamicamente os dados do tipo json, para keyby e somar com a coluna fornecida, em meu mapFunction, eu converto json para a classe de caso, mas o fluxo de resultado não obtém o compilador na função keyBy, obteve o erro. Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: This type (GenericType<java.lang.Object>) cannot be used as key.Meu código como isso
//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: _*)
}}
como posso converter json para classe de caso ou tupla?
Respostas
Na verdade, parecia que a exceção
org.apache.flink.api.common.InvalidProgramException:
This type (GenericType<Test>) cannot be used as key
permanece mesmo para a classe de caso comum (não gerado por reflexão)
case class Test(a: String, b: Int, c: String, d: Long)
O primeiro problema é que esta classe de caso não é um POJO
https://ci.apache.org/projects/flink/flink-docs-stable/dev/types_serialization.html#pojos
POJOs
As classes Java e Scala são tratadas pelo Flink como um tipo de dados POJO especial se cumprirem os seguintes requisitos:
A aula deve ser pública.
Deve ter um construtor público sem argumentos (construtor padrão).
Todos os campos são públicos ou devem ser acessíveis por meio das funções getter e setter. Para um campo denominado foo, os métodos getter e setter devem ser nomeados getFoo () e setFoo ().
O tipo de um campo deve ser compatível com um serializador registrado.
Então você deve substituir
case class Test(a: String, b: Int, c: String, d: Long)
com
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)
}
}
O segundo problema pode ser que o Flink não permite POJOs de classe interna que não são classes internas estáticas, mas a caixa de ferramentas reflexiva gera uma classe local aninhada em um método
Regras para tipos POJO
O Flink reconhece um tipo de dados como um tipo POJO (e permite a referência de campo "por nome") se as seguintes condições forem atendidas:
- A classe é pública e autônoma (sem classe interna não estática)
- A classe tem um construtor público sem argumento
- Todos os campos não estáticos e não transitórios na classe (e todas as superclasses) são públicos (e não finais) ou têm um método getter- e setter- público que segue as convenções de nomenclatura dos beans Java para getters e setters.
Aqui está a versão descompilada do código gerado pela caixa de ferramentas
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();
}
...
}
O código descompilado completo:
https://gist.github.com/DmytroMitin/f1554ad833ea1bb9eb97947ae872d220
Portanto, é possível que, se for realmente necessário gerar uma classe para o Flink, ela seja gerada manualmente em vez da caixa de ferramentas
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/
Mas o código com uma classe gerada manualmente
https://gist.github.com/DmytroMitin/e33cd244b37f9b33b67f7ac3e6609d39
ainda joga This type (GenericType<java.lang.Object>) cannot be used as key.
Acho que a razão para isso é a seguinte (e esta é a terceira questão ).
O código com classe de caso comum (não gerado) parece funcionar
https://gist.github.com/DmytroMitin/af426d4578dd5e76c9e0d344e6f079ce
Mas se substituirmos o tipo Testpor Any, ele lançaThis type (GenericType<java.lang.Object>) cannot be used as key
https://gist.github.com/DmytroMitin/a23e45a546790630e838e60c7206adcd
E com reflexão, não podemos retornar nada, mas Any.
Agora estou criando TypeInformation[Test]dentro do meu código gerado, isso parece corrigir, This type (GenericType<java.lang.Object>) cannot be used as keymas agora tenho
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
Resolvi o problema com InvalidProgramException: UTF-8 is not serializablecampos de anotação de MapFunccom@transient
https://gist.github.com/DmytroMitin/f2f859273075370c4687a30e0c3a2431
Na verdade, parecia que se criarmos TypeInformationdentro do código gerado, a caixa de ferramentas é 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: _*)
}
}