链接到 NiFi 连接器

Question

我需要一些帮助，使用Scala代码将数据从输出NiFi端口传输到Flink。

我被困在.addSource()功能上。它要求其他参数([OUT](，但是当我提供它们时，我不断收到错误。Scala 代码和错误消息如下。

package flinkTest
import java.nio.charset.{Charset, StandardCharsets}
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.connectors.nifi.NiFiSource
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.connectors.nifi.NiFiDataPacket
import org.apache.nifi.remote.client.{SiteToSiteClient, SiteToSiteClientConfig}
object NifiFlow {
def main(): Unit = {
// get the execution environment
val env: StreamExecutionEnvironment = 
StreamExecutionEnvironment.getExecutionEnvironment
// get input data by connecting to NiFi
val clientConfig: SiteToSiteClientConfig = new SiteToSiteClient.Builder()
.url("http://localhost:8080/nifi")
.portName("Data to flink")
.requestBatchCount(2)
.buildConfig()
val nifiSource: SourceFunction[NiFiDataPacket] = new NiFiSource(clientConfig)

这是一块

val streamSource: DataStream[NiFiDataPacket] = 
env.addSource(nifiSource).setParallelism(2)

以及更多代码

val dataStream = streamSource.map(dataPacket => new String(dataPacket.getContent, StandardCharsets.UTF_8))
dataStream.print()
env.execute()
}
}

1( 带 [输出]

Error:(28, 76) value nifiSource of type org.apache.flink.streaming.api.functions.source.SourceFunction[org.apache.flink.streaming.connectors.nifi.NiFiDataPacket] does not take type parameters.
val streamSource: DataStream[NiFiDataPacket] = env.addSource(nifiSource[NiFiDataPacket]).setParallelism(2)

2( 没有 [输出]

Error:(28, 66) type mismatch;
found   : org.apache.flink.streaming.api.functions.source.SourceFunction[org.apache.flink.streaming.connectors.nifi.NiFiDataPacket]
required: org.apache.flink.streaming.api.function.source.SourceFunction[?]
val streamSource: DataStream[NiFiDataPacket] = env.addSource(nifiSource).setParallelism(2)

这里取了一个例子，并重写为Scala。

我将不胜感激任何建议。

UPD2

package flinkTest
import org.apache.nifi.remote.client.{SiteToSiteClient, SiteToSiteClientConfig}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.nifi._
object NifiFlow {
def main(): Unit = {
// get the execution environment
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
// get input data by connecting to NiFi
val clientConfig: SiteToSiteClientConfig = new SiteToSiteClient.Builder()
.url("http://localhost:8080/nifi")
.portName("Data to flink")
.requestBatchCount(2)
.buildConfig()
val nifiSource = new NiFiSource(clientConfig)
val streamSource: DataStream[String] = env
.addSource(nifiSource)
.map(x => x.getAttributes().toString)
env.execute()
}
}

错误

Connected to the target VM, address: '127.0.0.1:41218', transport: 'socket'
Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: Interfaces and abstract classes are not valid types: interface org.apache.flink.streaming.connectors.nifi.NiFiDataPacket
at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:871)
at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:863)
at org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfoWithTypeHierarchy(TypeExtractor.java:406)
at org.apache.flink.api.java.typeutils.TypeExtractor.privateCreateTypeInfo(TypeExtractor.java:197)
at org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfo(TypeExtractor.java:184)
at flinkTest.NifiFlow$.main(NiFiFlow.scala:23)

Answer 1

有一个特殊的 scala 执行环境实现

org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

只是使用它而不是org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

Answer 2

env.addSource(nifiSource)仅适用于先前的设置env.getJavaEnv.getConfig.disableClosureCleaner()

也许，这个开源项目中的 scala 源代码应该更新一点(位于 flink-scala_2.11...jar 中(。