连接到NiFi连接器

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

我卡在.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）使用[OUT]

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）在不[OUT]

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)

示例采取了here并重写为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() 
    } 
}

ERROR

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)

来源

2017-07-17 Anna Vergeles

有斯卡拉专项实施的执行环境

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

只是用它来代替org.apache.flink.streaming.api.environment.StreamExecutionEnvironment

件

来源

2017-07-18 10:15:49 daggett

没有帮助。它以某种方式不能转换类型类。错误：（37，38）类型不匹配; 找到：org.apache.flink.streaming.api.functions.source.SourceFunction [** org.apache.flink.streaming.connectors.nifi.NiFiDataPacket **] 必需：org.apache.flink.streaming.api。 function.source.SourceFunction [**？**] val streamSource = env.addSource（nifiSource） –

你可以附上一个完整的文件，修改后的代码和完整的错误？ – daggett

更新了最初的帖子。代码太长，无法粘贴到评论中。 –

env.addSource(nifiSource)作品只有事先设定env.getJavaEnv.getConfig.disableClosureCleaner()

也许，在这个开源项目来源阶应更新位（位于弗林克-scala_2.11 ...罐子）。

来源

2017-07-21 09:54:30

连接到NiFi连接器

回答

相关问题