連接到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連接器

回答

相關問題