json in flink map dynamisch analysieren

Question

Ich verwende flink, um Daten vom Typ json dynamisch zu analysieren, um sie mit der angegebenen Spalte zu keybyen und zu summieren. In meiner mapFunction konvertiere ich json in eine case-Klasse, aber der Ergebnisstrom erhält keinen Compiler in der keyBy-Funktion und hat den Fehler Exception in thread "main" org.apache.flink.api.common.InvalidProgramException: This type (GenericType<java.lang.Object>) cannot be used as key..my code so was

//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: _*) 
}}

Wie kann ich json in case class oder tuple konvertieren?

Answer 1

Eigentlich schien es, dass die Ausnahme

org.apache.flink.api.common.InvalidProgramException: 
This type (GenericType<Test>) cannot be used as key 

bleibt auch für gewöhnliche Fallklassen (nicht durch Reflexion erzeugt)

case class Test(a: String, b: Int, c: String, d: Long)

Das erste Problem ist, dass diese Fallklasse kein POJO ist

https://ci.apache.org/projects/flink/flink-docs-stable/dev/types_serialization.html#pojos

POJOs

Java- und Scala-Klassen werden von Flink als spezieller POJO-Datentyp behandelt, wenn sie die folgenden Anforderungen erfüllen:

• Die Klasse muss öffentlich sein.

• Es muss einen öffentlichen Konstruktor ohne Argumente haben (Standardkonstruktor).

• Alle Felder sind entweder öffentlich oder müssen über Getter- und Setter-Funktionen zugänglich sein. Für ein Feld namens foo müssen die Methoden getter und setter getFoo () und setFoo () heißen.

• Der Feldtyp muss von einem registrierten Serializer unterstützt werden.

Also solltest du ersetzen

case class Test(a: String, b: Int, c: String, d: Long)

mit

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)
  }
}

Das zweite Problem kann sein, dass Flink keine POJOs der inneren Klasse zulässt, die keine statischen inneren Klassen sind, sondern die reflektierende Toolbox eine lokale Klasse generiert, die in einer Methode verschachtelt ist

https://ci.apache.org/projects/flink/flink-docs-stable/dev/types_serialization.html#rules-for-pojo-types

Regeln für POJO-Typen

Flink erkennt einen Datentyp als POJO-Typ (und ermöglicht die Feldreferenzierung nach Namen), wenn die folgenden Bedingungen erfüllt sind:

• Die Klasse ist öffentlich und eigenständig (keine nicht statische innere Klasse)
• Die Klasse verfügt über einen öffentlichen Konstruktor ohne Argumente
• Alle nicht statischen, nicht vorübergehenden Felder in der Klasse (und alle Superklassen) sind entweder öffentlich (und nicht endgültig) oder verfügen über eine öffentliche Getter- und eine Setter-Methode, die den Java-Beans-Namenskonventionen für Getter und Setter folgt.

Hier ist eine dekompilierte Version des von der Toolbox generierten Codes

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();
   }

   ...
}

Der vollständig dekompilierte Code:

https://gist.github.com/DmytroMitin/f1554ad833ea1bb9eb97947ae872d220

Wenn es also wirklich notwendig ist, eine Klasse für Flink zu generieren, sollte diese manuell und nicht über eine Toolbox generiert werden

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/53976254/how-to-eval-code-that-uses-interfacestability-annotation-that-fails-with-illeg

https://stackoverflow.com/questions/2946338/how-do-i-programmatically-compile-and-instantiate-a-java-class

https://stackoverflow.com/questions/56922911/dynamic-compilation-of-multiple-scala-classes-at-runtime

https://stackoverflow.com/questions/60783153/tensorflow-in-scala-reflection/

Aber der Code mit einer Klasse wird manuell generiert

https://gist.github.com/DmytroMitin/e33cd244b37f9b33b67f7ac3e6609d39

wirft immer noch This type (GenericType<java.lang.Object>) cannot be used as key.

Ich denke, der Grund dafür ist der folgende (und dies ist das dritte Problem ).

Der Code mit der normalen Fallklasse (nicht generiert) scheint zu funktionieren

https://gist.github.com/DmytroMitin/af426d4578dd5e76c9e0d344e6f079ce

Aber wenn wir den Typ Testdurch ersetzen, wird Anyer geworfenThis type (GenericType<java.lang.Object>) cannot be used as key

https://gist.github.com/DmytroMitin/a23e45a546790630e838e60c7206adcd

Und mit Nachdenken können wir nichts anderes zurückgeben Any.

---

Jetzt erstelle ich TypeInformation[Test]in meinem generierten Code, das scheint zu beheben, This type (GenericType<java.lang.Object>) cannot be used as keyaber jetzt habe ich

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

---

Ich habe das Problem mit dem InvalidProgramException: UTF-8 is not serializableKommentieren von Feldern MapFuncmit behoben@transient

https://gist.github.com/DmytroMitin/f2f859273075370c4687a30e0c3a2431

---

Tatsächlich schien es, dass die TypeInformationToolbox ausreicht , wenn wir generierten Code erstellen

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: _*)
  }
}