spark kafka consumer 消费数据的二种方式

张映 发表于 2019-04-28

分类目录: hadoop/spark/scala

标签:, ,

如果实时的从kafka取数据,通过spark入hdfs,会产生很多的task,在hdfs上会产生非常多的小文件。浪费硬盘空间不说,在用spark进行数据分析的时间,非常耗spark节点的内存。

最好不要实时的入库,间断去运行。这样会尽量减少小文件的产生。但是不能根本上解决小问题,最终还是通过CombineFileInputFormat来解决,这个后面的文章,会单独说。

1,采crontab的方式

这种方式也是最容易想到的一种方式,spark-submit提交后,启动sparksession,启动kafka consumer,消费数据。这儿有一点要注意,数据入hdfs后,关闭sparksession,kafka consumer,这样可以节约系统资源。

2,采用akka包,处理方式,根crontab差不多(推荐)

pom.xml加载包,注意和当前scala版本要对的上,不然打包时会报错

<dependency>
 <groupId>com.typesafe.akka</groupId>
 <artifactId>akka-actor_2.11</artifactId>
 <version>2.5.9</version>
</dependency>

例子:

object test {
    def main(args: Array[String]): Unit = {
        。。。。。。。。。。。。。。。。省略。。。。。。。。。。。。。。。
        val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](pros)
        /*这里填写主题名称*/
        consumer.subscribe(util.Arrays.asList(table))
        val system = akka.actor.ActorSystem("system")
        system.scheduler.schedule(0 seconds, 180 seconds)(taskerPc.saveData(args,consumer))
    }

    object taskerPc {
        def saveData(args: Array[String],consumer: KafkaConsumer[String,String]): Unit = {
            。。。。。。。。。。。。。。。。省略。。。。。。。。。。。。。。。
            /*
            *
            * spark.sql.warehouse.dir hdfs://主数据节点别名或者ip:post指定单个主机/
            * */
            val spark = new sql.SparkSession.Builder()
                    .config("spark.sql.warehouse.dir", func.cnf("spark.sql.warehouse.dir"))
                    .enableHiveSupport()
                    .appName(table)
                    .getOrCreate()

            val records: ConsumerRecords[String, String] = consumer.poll(Duration.ofSeconds(3))
            。。。。。。。。。。。。。。。。省略。。。。。。。。。。。。。。。
        }
    }
}

scala main函数中,开起了一个kafka consumer,会每隔180秒,去调用函数saveData,这种方式,consumer是不能关闭的,一关闭就无法消费topic里面的数据了。如果把val consumer: KafkaConsumer[String, String] = new KafkaConsumer[String, String](pros),放到了saveData中,就要关闭consumer,不然就会出现Attempt to heartbeat failed since group is rebalancing问题。



转载请注明
作者:海底苍鹰
地址:http://blog.51yip.com/hadoop/2129.html