Scala spark kafka代码函数方法

Question

下面是Scala中的代码。我正在使用spark sql从hadoop中提取数据，对结果执行一些分组，序列化它，然后将消息写给Kafka。

我已经写了代码--但我想用函数的方式来写。我是否应该创建一个具有“get categories”函数的新类来从Hadoop中获取类别？我不知道如何处理这件事。

这是代码

class ExtractProcessor {
  def process(): Unit = {

  implicit val formats = DefaultFormats

  val spark = SparkSession.builder().appName("test app").getOrCreate()

  try {
     val df = spark.sql("SELECT DISTINCT SUBCAT_CODE, SUBCAT_NAME, CAT_CODE, CAT_NAME " +
    "FROM CATEGORY_HIERARCHY " +
    "ORDER BY CAT_CODE, SUBCAT_CODE ")

     val result = df.collect().groupBy(row => (row(2), row(3)))
     val categories = result.map(cat =>
                    category(cat._1._1.toString(), cat._1._2.toString(),
                      cat._2.map(subcat =>
                      subcategory(subcat(0).toString(), subcat(1).toString())).toList))

     val jsonMessage = write(categories)
     val kafkaKey = java.security.MessageDigest.getInstance("SHA-1").digest(jsonMessage.getBytes("UTF-8")).map("%02x".format(_)).mkString.toString()
     val key = write(kafkaKey)

     Logger.log.info(s"Json Message: ${jsonMessage}")
     Logger.log.info(s"Kafka Key: ${key}")

     KafkaUtil.apply.send(key, jsonMessage, "testTopic")      
}

class KafkaUtil {
  def send(key: String, message: String, topicName: String): Unit = {
  val properties = new Properties()
  properties.put("bootstrap.servers", "localhost:9092")
  properties.put("client.id", "test publisher")
  properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
  val producer = new KafkaProducer[String, String](properties)

  try {

    val record = new ProducerRecord[String, String](topicName, key, message)
    producer.send(record)
  }
  finally {
    producer.close()
    Logger.log.info("Kafka producer closed...")
  }
 }
}

object KafkaUtil {
  def apply: KafkaUtil = {
  new KafkaUtil
 }
}

提前谢谢你，苏约格

Answer 1

您的代码包括：1）将数据加载到spark df中；2）处理数据；3）创建json消息；4）向kafka发送json消息

单元测试对于测试纯函数是很好的。您可以将步骤2)提取到具有类似def getCategories(df:DataFrame):seq[Category]签名的方法中，并用测试覆盖它。

步骤3)也可以包含在单元测试中，如果您觉得它容易出错

步骤1)和4)将由端到端测试涵盖

顺便说一句，val result=df.collect().groupby(row=>(row(2)，row(3)))效率很低。最好将其替换为val result=df.groupby(row=>(row(2)，row(3))).collect