IntelliJ не может разрешить символ Подписаться kafka - PullRequest
Новая
       6

IntelliJ не может разрешить символ Подписаться kafka

0 голосов
/ 22 мая 2018

У меня следующий проект в IntelliJ, проблема в том, что Subscribe внутри KafkaUtils.createDirectStream выглядит красным, и он выдает Cannot resolve symbol Subscribe, но я добавил все библиотеки kafka-spark: 

    import org.apache.spark.streaming.{Seconds, StreamingContext}
    import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
    import org.apache.spark.streaming.kafka010.KafkaUtils
    import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent

  def startMetaInfoSubscriber(ssc: StreamingContext, kafkaParams: Map[String, Object], metaInfoTopic: String) {
    // Set a unique Kafka group identifier to metaInformationStream (each stream requires a unique group ID)
    val metaInformationKafkaParamas = kafkaParams ++ Map[String, Object]("group.id" -> RandomStringUtils.randomAlphabetic(10).toUpperCase)

    KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](metaInfoTopic, metaInformationKafkaParamas)
    ).foreachRDD(metaInfoRDD =>
      if (!metaInfoRDD.isEmpty()) {
        println("Saving MetaInformation")
        metaInfoRDD
//        metaInfoRDD.write.mode("append").format("com.databricks.spark.csv").save(s"hdfs://172.16.8.162:8020/user/sparkload/assetgroup/prueba-kafka")
      } else {
        println("There is not any message for topic 'tu-topic'")
      }
    )
  }

И следующий мой pom.xml: 

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<scala.version>2.11.8</scala.version>
<spark.version>2.3.0</spark.version>
<src.dir>src/main/scala</src.dir>

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core_2.11 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka_2.11 -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
    <version>${spark.version}</version>
</dependency>

Когда я пытаюсь скомпилировать, у меня появляется следующая ошибка: 

[ERROR] C:\Users\agomez\Desktop\spark-base\spark-kafka-tfm\src\main\scala\spark_load\EjemploApp.scala:90: error: overloaded method value Subscribe with alternatives:
[ERROR]   (topics: java.util.Collection[String],kafkaParams: java.util.Map[String,Object])org.apache.spark.streaming.kafka010.ConsumerStrategy[String,String] <and>
[ERROR]   (topics: java.util.Collection[String],kafkaParams: java.util.Map[String,Object],offsets: java.util.Map[org.apache.kafka.common.TopicPartition,java.lang.Long])org.apache.spark.streaming.kafka010.ConsumerStrategy[String,String] <and>
[ERROR]   (topics: Iterable[String],kafkaParams: scala.collection.Map[String,Object])org.apache.spark.streaming.kafka010.ConsumerStrategy[String,String] <and>
[ERROR]   (topics: Iterable[String],kafkaParams: scala.collection.Map[String,Object],offsets: scala.collection.Map[org.apache.kafka.common.TopicPartition,scala.Long])org.apache.spark.streaming.kafka010.ConsumerStrategy[String,String]
[ERROR]  cannot be applied to (String, scala.collection.immutable.Map[String,Object])
[ERROR]       Subscribe[String, String](metaInfoTopic, metaInformationKafkaParamas)
[ERROR]                ^
[ERROR] one error found

1 Ответ

0 голосов
/ 22 мая 2018

Я думаю, что первым параметром Subscribe() должен быть Набор тем

Так что вам нужно передать несколько тем как Seq[Strings] или Array[Strings].Если у вас есть отдельная тема, просто передайте ее как Seq(metaInfoTopic)

KafkaUtils.createDirectStream[String, String](
    ssc,
    PreferConsistent,
    Subscribe[String, String](Seq(metaInfoTopic), metaInformationKafkaParamas)
)

Надеюсь, это поможет!

...