Kafka工具類（Scala）

1、配置文件config.properties

# Kafka配置
kafka.broker.list = hadoop300:9092,hadoop301:9092,hadoop302:9092
# Redis配置
redis.host=hadoop300
redis.port=6379

 

2、讀取Properties

package com.duoduo.realtime.utils

import java.io.InputStreamReader
import java.util.Properties

/**
 * Author z
 * Date 2020-08-27 10:04:21
 */
object PropertiesUtil {
  def main(args: Array[String]): Unit = {
    val properties: Properties = PropertiesUtil.load("config.properties")
    
    println(properties.getProperty("kafka.broker.list"))
  }
  def load(propertiesName: String) = {
    val p=new Properties()
      p.load(new InputStreamReader(
        Thread.currentThread().getContextClassLoader
          .getResourceAsStream(propertiesName)
        , "UTF-8"))
    p
  }
}

 

3、POM文件依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>dw-stream</artifactId>
        <groupId>com.duoduo</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>dw-realtime</artifactId>
    <properties>
        <spark.version>2.4.0</spark.version>
        <scala.version>2.11.8</scala.version>
        <kafka.version>1.0.0</kafka.version>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.56</version>
        </dependency>



        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>

        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>2.9.0</version>
        </dependency>


        <dependency>
            <groupId>org.apache.phoenix</groupId>
            <artifactId>phoenix-spark</artifactId>
            <version>4.14.2-HBase-1.3</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>io.searchbox</groupId>
            <artifactId>jest</artifactId>
            <version>5.3.3</version>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-api</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>net.java.dev.jna</groupId>
            <artifactId>jna</artifactId>
            <version>4.5.2</version>
        </dependency>

        <dependency>
            <groupId>org.codehaus.janino</groupId>
            <artifactId>commons-compiler</artifactId>
            <version>2.7.8</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <!-- 該插件用於將Scala代碼編譯成class文件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.4.6</version>
                <executions>
                    <execution>
                        <!-- 聲明綁定到maven的compile階段 -->
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

 

4、工具類

package com.duoduo.realtime.utils

import java.util.Properties

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, ConsumerStrategy, KafkaUtils, LocationStrategies}

/**
 * Author z
 * Date 2020-08-27 10:02:21
 */
object KafkaUtil {
  private val properties: Properties = PropertiesUtil.load("config.properties")
  val broker_list = properties.getProperty("kafka.broker.list")
  var kafkaParam = collection.mutable.Map(
    "bootstrap.servers" -> broker_list, //用於初始化鏈接到集群的地址
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    //用於標識這個消費者屬於哪個消費團體
    "group.id" -> "gmall_consumer_group",
    //如果沒有初始化偏移量或者當前的偏移量不存在任何服務器上，可以使用這個配置屬性
    //可以使用這個配置，latest自動重置偏移量為最新的偏移量
    "auto.offset.reset" -> "latest",
    //如果是true，則這個消費者的偏移量會在后台自動提交,但是kafka宕機容易丟失數據
    //如果是false，會需要手動維護kafka偏移量
    "enable.auto.commit" -> (false: java.lang.Boolean)
  )
  
  def getKafkaStream(topic: String, ssc: StreamingContext)
  : InputDStream[ConsumerRecord[String, String]] = {
    val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils
      .createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParam)
      )
    dStream
  }
  
  def getKafkaStream(topic: String, ssc: StreamingContext, groupid: String)
  : InputDStream[ConsumerRecord[String, String]] = {
    kafkaParam("group.id") = groupid
    val dStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils
      .createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](Array(topic), kafkaParam)
      )
    dStream
  }
  
  def getKafkaStream(topic: String,
                     ssc: StreamingContext,
                     offsets: Map[TopicPartition, Long]
                     , groupid: String)
  : InputDStream[ConsumerRecord[String, String]] = {
    kafkaParam("group.id") = groupid
    val dStream: ConsumerStrategy[String, String] = ConsumerStrategies
      .Subscribe[String, String](Array(topic), kafkaParam, offsets)
    KafkaUtils.createDirectStream[String, String](
      ssc,
      LocationStrategies.PreferConsistent,
      dStream
    )
  }
}