暂时未有相关云产品技术能力~
问题似乎是SBT和Maven配置文件不能很好地协同工作。Flink POM将Scala版本(2.10,2.11,...)称为变量,其中一些在构建配置文件中定义。 SBT未正确评估配置文件,因此包装无法正常工作。
使用以下build.sbt:
organization := "pl.japila.kafka"
scalaVersion := "2.11.7"
libraryDependencies += "org.apache.flink" % "flink-connector-kafka" % "0.9.0" exclude("org.apache.kafka", "kafka_${scala.binary.version}")
libraryDependencies += "org.apache.kafka" %% "kafka" % "0.8.2.1"
运行如下:
import org.apache.flink.streaming.api.environment._
import org.apache.flink.streaming.connectors.kafka
import org.apache.flink.streaming.connectors.kafka.api._
import org.apache.flink.streaming.util.serialization._
object TestKafka {
def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val stream = env
.addSource(new KafkaSource[String]("localhost:2181", "test", new SimpleStringSchema))
.print
}
}
输出:
[kafka-flink]> run
[info] Running TestKafka
log4j:WARN No appenders could be found for logger (org.apache.flink.streaming.api.graph.StreamGraph).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
[success] Total time: 0 s, completed Jul 15, 2015 9:29:31 AM
Flink是统一流和批处理的框架。由于并行任务之间的流水线数据传输,Flink运行时本身支持这两个域。记录被立即地从生产任务发送到接收任务,可以使用阻塞数据传输选择性地执行批处理作业。Spark是一个支持批处理和流处理的框架,但和Flink在内部方面有所不同。对于流式传输,两种系统都遵循非常不同的方法(小批量与流式传输),这使得它们适用于不同类型的应用。