Kafka Spark Scala Cassandra対応バージョン

Question

Apache Kafka、Saprk、Scala、Cassandraを使用してアプリケーションを作成しようとしています。 しかし、私はこれらのツールの正しいバージョンを入手する際に多くの問題に直面しています。

使用するバージョンを教えていただけますか？ここで

事前のおかげで..

Answer 1

は、我々が使用したライブラリのバージョンのリストです：

<dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-core_2.11</artifactId> 
     <version>2.1.1</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.spark</groupId> 
     <artifactId>spark-streaming-kafka-0-10_2.11</artifactId> 
     <version>2.1.1</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.kafka</groupId> 
     <artifactId>kafka-clients</artifactId> 
     <version>0.10.2.0</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.kafka</groupId> 
     <artifactId>kafka_2.11</artifactId> 
     <version>0.10.2.1</version> 
    </dependency> 
    <dependency> 
     <groupId>org.apache.cassandra</groupId> 
     <artifactId>apache-cassandra</artifactId> 
     <version>3.10</version> 
    </dependency> 
    <dependency> 
     <groupId>com.datastax.spark</groupId> 
     <artifactId>spark-cassandra-connector_2.11</artifactId> 
     <version>2.0.2</version> 
    </dependency> 

あなたは、互換性と直面する主な問題は、異なるScalaのバージョン（2.10です。 *または2.11。*）。それを見て、すべての依存関係が同じスカラーバージョンを使用していることを確認する必要があります。私はあなたがどこでも同じスカラーのバージョンを気にするのであれば、最新のすべてのバージョンを疑いなく更新できると思います。ここで

もスタートをお届けしますコードサンプルです：コネクタで

 public static void main(String[] args) throws InterruptedException { 
      JavaStreamingContext jssc = new JavaStreamingContext(getSparkConfiguration(), Durations.seconds(5)); 

      JavaInputDStream<ConsumerRecord<String, LoggingEvent>> messages = 
      KafkaUtils.createDirectStream(
        jssc, 
        LocationStrategies.PreferConsistent(), 
        ConsumerStrategies.<String, LoggingEvent>Subscribe(Arrays.asList("some_topic"), getKafkaParams("localhost:9092", "some_logging_group)) 
      ); 

      JavaDStream<LoggingEvent> loggingRecords = messages.map(
      (Function<ConsumerRecord<String, LoggingEvent>, LoggingEvent>) message -> message.value() 
    ); 

      CassandraStreamingJavaUtil.javaFunctions(loggingRecords).writerBuilder("some_space", "some_table", 
        CassandraJavaUtil.mapToRow(LoggingEvent.class)).saveToCassandra(); 

      jssc.start(); 
      jssc.awaitTermination(); 
} 

マッピングは、テーブルの列とクラスのマッピングフィールドで行われます。

セットアップ用に、アーカイブ用のバージョンと配布バージョンをライブラリの依存関係リストと同じにしました。

Answer 2

場合、あなたはSBTバージョン

libraryDependencies ++= { 

    val sparkV = "2.1.0" 
    val cassandraV = "2.0.0-M3" 

    Seq(
    "org.apache.spark"  %% "spark-core" % sparkV, 
    "org.apache.spark"  %% "spark-streaming" % sparkV, 
    "org.apache.spark"  %% "spark-streaming-kafka-0-10" % sparkV, 
    "org.apache.spark"  %% "spark-sql-kafka-0-10" % sparkV, 
    "org.apache.spark"  %% "spark-sql" % sparkV, 
    "org.apache.spark"  %% "spark-hive" % sparkV, 
    "com.datastax.spark" %% "spark-cassandra-connector" % cassandraV, 
    "com.datastax.cassandra" % "cassandra-driver-core" % "3.2.0", 
    "org.rogach"   %% "scallop" % "2.1.2" 
) 

}