我们从kafka获取记录，在spark streaming中从kafka获取cardnumber，并从memsql记录中执行kafka cardnumber比较，通过分组cardnumber选择计数和cardnumber。但是在星火流中计数的方式不正确
例如，当我们执行查询时，它在memsql命令提示符中给出下面的输出

memsql> select card_number,count(*) from cardnumberalert5 where 
inserted_time <= now() and inserted_time >= NOW() - INTERVAL 10 MINUTE group 
by card_number;
+------------------+----------+
| card_number      | count(*) |
+------------------+----------+
| 4556655960290527 |        2 |
| 6011255715328120 |        4 |
| 4532133676538232 |        2 |
| 6011614607071620 |        2 |
| 4024007117099605 |        2 |
| 347138718258304  |        4 |
+------------------+----------+

我们注意到在Spark流中，计数正在被分配
例如，从memsql命令提示符执行时的memsql输出

+------------------+----------+
| card_number      | count(*) |
+------------------+----------+
| 4556655960290527 |        2 |

当在spark streaming中执行相同的sql时，它将输出打印为

RECORDS FOUNDS****************************************
CARDNUMBER KAFKA ############### 4024007117099605
CARDNUMBER MEMSQL ############### 4556655960290527
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 4556655960290527
COUNT MEMSQL ############### 1

这里的计数显示为2，但我们得到了2个记录的卡号与计数1
在spark streaming中打印输出

RECORDS FOUNDS****************************************
CARDNUMBER KAFKA ############### 4024007117099605
CARDNUMBER MEMSQL ############### 4556655960290527
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 6011255715328120
COUNT MEMSQL ############### 2
CARDNUMBER MEMSQL ############### 4532133676538232
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 6011614607071620
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 4024007117099605
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 347138718258304
COUNT MEMSQL ############### 2
CARDNUMBER MEMSQL ############### 4556655960290527
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 6011255715328120
COUNT MEMSQL ############### 2
CARDNUMBER MEMSQL ############### 4532133676538232
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 6011614607071620
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 4024007117099605
COUNT MEMSQL ############### 1
CARDNUMBER MEMSQL ############### 347138718258304
COUNT MEMSQL ############### 2

spark流媒体程序

class SparkKafkaConsumer11(val ssc : StreamingContext,val sc : SparkContext,val spark : org.apache.spark.sql.SparkSession, val topics : Array[String], val kafkaParam : scala.collection.immutable.Map[String,Object]) {

 val stream = KafkaUtils.createDirectStream[String, String](
            ssc,
            PreferConsistent,
            Subscribe[String, String](topics, kafkaParam)
          )

  val recordStream = stream.map(record => (record.value)) // Take the value only from the key,value pair for processing

   recordStream.foreachRDD{rdd =>

val brokers = "174.24.154.244:9092" // Specify the BROKER
val props = new HashMap[String, Object]()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.CLIENT_ID_CONFIG,"SparkKafkaConsumer__11")
val producer = new KafkaProducer[String,String](props)

val result = spark.read
            .format("com.memsql.spark.connector")
            .options(Map("query" -> ("select card_number,count(*) from cardnumberalert5 where inserted_time <= now() and inserted_time >= NOW() - INTERVAL 10 MINUTE group by card_number"),"database" -> "fraud"))
            .load()

val record = rdd.map(line => line.split("\\|")) //Split the record and create a array of it.

 record.collect().foreach{recordRDD =>
    val now1 = System.currentTimeMillis

    val now = new java.sql.Timestamp(now1)
    val cardnumber_kafka = recordRDD(13).toString
    val sessionID = recordRDD(1).toString
    println("RECORDS FOUNDS****************************************")
    println("CARDNUMBER KAFKA ############### "+cardnumber_kafka)

    result.collect().foreach{t => 

      val resm1 = t.getAs[String]("card_number")
      println("CARDNUMBER MEMSQL ############### "+resm1)
      val resm2 = t.getAs[Long]("count(*)")
      println("COUNT MEMSQL ############### "+resm2)

      if(resm1.equals(cardnumber_kafka)){
        if(resm2 > 2){
          println("INSIDE IF CONDITION FOR MORE THAN 3 COUNT"+now)
          val messageToKafka = "---- THIRD OR MORE OCCURANCE  ---- "+cardnumber_kafka
          val message=new ProducerRecord[String, String]("output1",0,sessionID,messageToKafka)
          try {
            producer.send(message)

          } catch {
              case e: Exception =>
              e.printStackTrace
              System.exit(1)
          }
        }
      }

    }

}

producer.close()

}

}

不知道如何解决它，任何建议或帮助是高度赞赏
提前谢谢