Caused by: java.io.NotSerializableException: org.apache.parquet.schema.MessageType

系统管理员 2022-10-14 05:28 95阅读 0赞

目前没解决，本地代码可以运行，打包集群运行也没问题，可能是zeppelin哪里的依赖包冲突。

>  
> 
> scala任务代码
> 
>  
> 
>  
> 
>  
> 
> Head
> 
>    
> 
>    default 
> 
>  
> 
> %flink.conf
> 
> flink.execution.packages org.apache.flink:flink-connector-kafka\_2.11:1.11.2,com.alibaba:fastjson:1.2.60,org.apache.flink:flink-parquet\_2.11:1.11.2,org.apache.parquet:parquet-hadoop:1.11.1,org.apache.flink:flink-core:1.11.2
> 
> flink.execution.jars /wyyt/software/1.11.2-flink/flink-scala-opt-1.0-SNAPSHOT-shade.jar
> 
> FINISHED   
> 
> Took 0 sec. Last updated by anonymous at May 21 2021, 11:28:25 AM.
> 
>  
> 
> %flink.conf
> 
> READY   
> 
>  
> 
> %flink.conf
> 
> flink.execution.packages org.apache.flink:flink-connector-kafka\_2.11:1.12.2,com.alibaba:fastjson:1.2.60,org.apache.hadoop:hadoop-client:3.0.0,org.apache.flink:flink-parquet\_2.11:1.12.2,org.apache.hadoop:hadoop-common:3.0.0,org.apache.flink
> 
>     :flink-core:1.12.2
> 
> FINISHED   
> 
> Took 0 sec. Last updated by anonymous at May 20 2021, 4:06:51 PM. (outdated)
> 
>  
> 
> %flink.conf
> 
> flink.execution.packages org.apache.flink:flink-connector-kafka\_2.11:1.12.2,com.alibaba:fastjson:1.2.60,org.apache.flink:flink-parquet\_2.11:1.12.2,org.apache.flink:flink-core:1.12.2,org.apache.flink:flink-shaded-hadoop-2-uber:2.7.5-9.0
> 
> flink.execution.jars /wyyt/software/flink-1.12.2-2.11/flink-1.12.2/flink-hudi-1.0-SNAPSHOT-shade.jar
> 
> READY   
> 
>  
> 
> %flink
> 
> val data = benv.fromElements("hello world", "hello flink", "hello hadoop")
> 
> data.flatMap(line => line.split("\\\\s"))
> 
> .map(w => (w, 1))
> 
> .groupBy(0)
> 
> .sum(1)
> 
> .print()
> 
> data: org.apache.flink.api.scala.DataSet\[String\] = org.apache.flink.api.scala.DataSet@709f0986 (flink,1) (hadoop,1) (world,1) (hello,3)
> 
> [ FLINK JOB ][FLINK JOB]FINISHED   
> 
> Took 53 sec. Last updated by anonymous at May 21 2021, 11:29:19 AM.
> 
>  
> 
> %flink
> 
> READY   
> 
>  
> 
> import com.sjb.kafka.\{KafkaOutputFormat\}
> 
> import org.apache.commons.lang3.StringUtils
> 
> import org.apache.flink.api.common.io.FileInputFormat
> 
> import org.apache.flink.api.java.utils.ParameterTool
> 
> import org.apache.flink.api.scala.ExecutionEnvironment
> 
> import org.apache.flink.api.scala.\_
> 
> import org.apache.flink.core.fs.Path
> 
> import org.apache.flink.formats.parquet.ParquetRowInputFormat
> 
> import org.apache.flink.types.Row
> 
> import org.apache.hadoop.conf.Configuration
> 
> import org.apache.hadoop.fs.\{FileSystem, LocatedFileStatus, RemoteIterator\}
> 
> import org.apache.parquet.column.ColumnDescriptor
> 
> import org.apache.parquet.hadoop.ParquetFileReader
> 
> import org.apache.parquet.hadoop.metadata.ParquetMetadata
> 
> import org.apache.parquet.schema.MessageType
> 
> import scala.collection.JavaConversions.\_
> 
> benv.setParallelism(1)
> 
> val config: Configuration = new Configuration
> 
> val namenode: String = "hdfs://bi-524:8020"
> 
> config.set("fs.defaultFS", namenode)
> 
> //todo 写活的。
> 
> // val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
> 
> //todo 传入参数 --table xxx --database xxx --sinktopic xx --prod prod
> 
> // String databaseName = parameterTool.getRequired("database")
> 
> val databaseName: String = "test"
> 
> // String tableName = parameterTool.getRequired("table")
> 
> val tableName: String = "fs\_plt\_assure\_orders\_bak"
> 
> // String sinkTopic = parameterTool.getRequired("sinktopic")
> 
> val sinkTopic: String = "cep\_test"
> 
> // val patternParam: String = parameterTool.get("prod", "")
> 
> var path: String = ""
> 
> var serverProd: String = ""
> 
> serverProd ="dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092"
> 
> path = "hdfs://bi-524:8020/user/hive/warehouse/" + databaseName + ".db/"
> 
> val p: org.apache.hadoop.fs.Path = new org.apache.hadoop.fs.Path(path)
> 
> val fs: FileSystem = FileSystem.get(config)
> 
> print(fs.exists(p))
> 
> if (!fs.exists(p) || !fs.isDirectory(p)) \{
> 
> \}
> 
> var locatedFileStatusRemoteIterator: RemoteIterator\[LocatedFileStatus\] = null
> 
> var messageType: MessageType = null
> 
> val coStr: util.List\[String\] = new util.ArrayList\[String\]
> 
> val tableHdfsPathStr: org.apache.hadoop.fs.Path = new org.apache.hadoop.fs.Path(path + tableName + "///")
> 
> if (fs.isDirectory(tableHdfsPathStr)) \{
> 
> locatedFileStatusRemoteIterator = fs.listFiles(tableHdfsPathStr, true)
> 
> var flieName: org.apache.hadoop.fs.Path = null
> 
> //todo 循环取每个分片File的名字：000000\_0
> 
> if (locatedFileStatusRemoteIterator.hasNext) \{
> 
> flieName = locatedFileStatusRemoteIterator.next.getPath
> 
> if (flieName != null) \{
> 
> val parquetFileReader: ParquetMetadata = ParquetFileReader.readFooter(config, flieName)
> 
> messageType = parquetFileReader.getFileMetaData.getSchema
> 
> val columns: util.List\[ColumnDescriptor\] = messageType.getColumns
> 
> columns.foreach(x => \{
> 
> val aa: Array\[String\] = x.getPath
> 
> coStr.add(aa(0))
> 
> \})
> 
> \}
> 
> \}
> 
> \}
> 
> if (messageType != null) \{
> 
> val fileInputFormat: FileInputFormat\[Row\] = new ParquetRowInputFormat(new Path(path + tableName + "/"), messageType)
> 
> fileInputFormat.setNestedFileEnumeration(true)
> 
> val dataSet: DataSet\[Row\] = benv.createInput(fileInputFormat).setParallelism(1)
> 
> // val dataSet: DataSet\[Row\] = env.createInput(fileInputFormat).map(x=>Row.of(x)).setParallelism(1)
> 
> fileInputFormat.close()
> 
> val sinkDataSet = dataSet.map(line => \{
> 
> val json: JSONObject = new JSONObject
> 
> val lineSize: Int = line.getArity
> 
> for (i <- 0 until lineSize) \{
> 
> json.put(coStr.get(i), line.getField(i))
> 
> \}
> 
> if (json.containsKey("pk\_id")) \{
> 
> val pkName = json.getString("pk\_id")
> 
> if (pkName.contains("\\\\|")) \{
> 
> var pkId = ""
> 
> var pkValue = ""
> 
> var partitionerNum = 0
> 
> val split = pkId.split("\\\\|")
> 
> for (j <- 0 until split.length) \{
> 
> pkValue = pkValue + "|" + json.getString(split(j))
> 
> pkId = pkId + "|" + split(j)
> 
> partitionerNum += json.getString(split(j)).hashCode
> 
> \}
> 
> pkId = pkId.substring(1)
> 
> pkValue = pkValue.substring(1)
> 
> val hash = Math.abs(partitionerNum) % 3
> 
> //todo 设置为新的值
> 
> json.put("pk\_value", pkValue)
> 
> json.put("pk\_id", pkId)
> 
> json.put("pk\_hash", hash)
> 
> \} else \{
> 
> //todo 单个主键
> 
> val partitionerNum = json.getString(pkName).hashCode()
> 
> val hash = Math.abs(partitionerNum) % 3
> 
> json.put("pk\_value", json.getString(pkName))
> 
> json.put("pk\_hash", hash)
> 
> \}
> 
> \}
> 
> json.put("canal\_type", "INSERT")
> 
> json.put("canal\_ts", System.currentTimeMillis())
> 
> json.toJSONString
> 
> \})
> 
> sinkDataSet.output(KafkaOutputFormat.buildKafkaOutputFormat
> 
> .setBootstrapServers(serverProd)
> 
> .setTopic(sinkTopic)
> 
> .setAcks("1")
> 
> .setBatchSize("35000")
> 
> .setBufferMemory("33554432")
> 
> .setLingerMs("1")
> 
> .setRetries("1").finish)
> 
> .setParallelism(3).name("write kafka")
> 
> \}
> 
> import java.util import com.alibaba.fastjson.JSONObject import com.sjb.kafka.KafkaOutputFormat import org.apache.commons.lang3.StringUtils import org.apache.flink.api.common.io.FileInputFormat import org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.api.scala.ExecutionEnvironment import org.apache.flink.api.scala.\_ import org.apache.flink.core.fs.Path import org.apache.flink.formats.parquet.ParquetRowInputFormat import org.apache.flink.types.Row import org.apache.hadoop.conf.Configuration import org.apache.hadoop.fs.\{FileSystem, LocatedFileStatus, RemoteIterator\} import org.apache.parquet.column.ColumnDescriptor import org.apache.parquet.hadoop.ParquetFileReader import org.apache.parquet.hadoop.metadata.ParquetMetadata import org.apache.parquet.schema.MessageType import scala.collection.JavaConversions.\_ config: org.apache.hadoop.conf.Configuration = Configuration: core-default.xml, core-site.xml namenode: String = hdfs://bi-524:8020 databaseName: String = test tableName: String = fs\_plt\_assure\_orders\_bak sinkTopic: String = cep\_test path: String = "" serverProd: String = "" serverProd: String = dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092 path: String = hdfs://bi-524:8020/user/hive/warehouse/test.db/ p: org.apache.hadoop.fs.Path = hdfs://bi-524:8020/user/hive/warehouse/test.db fs: org.apache.hadoop.fs.FileSystem = DFS\[DFSClient\[clientName=DFSClient\_NONMAPREDUCE\_1839078995\_20, ugi=OpsUser (auth:SIMPLE)\]\] truelocatedFileStatusRemoteIterator: org.apache.hadoop.fs.RemoteIterator\[org.apache.hadoop.fs.LocatedFileStatus\] = null messageType: org.apache.parquet.schema.MessageType = null coStr: java.util.List\[String\] = \[\] tableHdfsPathStr: org.apache.hadoop.fs.Path = hdfs://bi-524:8020/user/hive/warehouse/test.db/fs\_plt\_assure\_orders\_bak/ warning: there was one deprecation warning; re-run with -deprecation for details org.apache.flink.api.common.InvalidProgramException: Task not serializable at org.apache.flink.api.scala.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:408) at org.apache.flink.api.scala.ClosureCleaner$.org$apache$flink$api$scala$ClosureCleaner$$clean(ClosureCleaner.scala:400) at org.apache.flink.api.scala.ClosureCleaner$.clean(ClosureCleaner.scala:168) at org.apache.flink.api.scala.DataSet.clean(DataSet.scala:125) at org.apache.flink.api.scala.DataSet$$anon$1.<init>(DataSet.scala:489) at org.apache.flink.api.scala.DataSet.map(DataSet.scala:488) ... 84 elided Caused by: java.io.NotSerializableException: org.apache.parquet.schema.MessageType at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548) at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509) at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432) at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178) at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348) at org.apache.flink.util.InstantiationUtil.serializeObject(InstantiationUtil.java:586) at org.apache.flink.api.scala.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:406) ... 89 more
> 
>

[FLINK JOB]: http://192.168.5.25:8081/#/job/61d96f7f49180d69d33f89e47d11eaa9