云服务器特惠 99/年起
  1. 腾讯云优惠网首页
  2. 腾讯云产品

消息队列 CKafka 版 Spark Streaming 接入 CKafka

• 腾讯云产品

Spark Streaming 是 Spark Core 的一个扩展，用于高吞吐且容错地处理持续性的数据，目前支持的外部输入有 Kafka、Flume、HDFS/S3、Kinesis、Twitter 和 TCP socket。

相关产品
消息队列 CKafka 版 产品页，含版本规格、价格与购买入口。

Spark Streaming 将连续数据抽象成 DStream（Discretized Stream），而 DStream 由一系列连续的 RDD（弹性分布式数据集）组成，每个 RDD 是一定时间间隔内产生的数据。使用函数对 DStream 进行处理其实即为对这些 RDD 进行处理。

使用 Spark Streaming 作为 Kafka 的数据输入时，可支持 Kafka 稳定版本与实验版本：

Kafka Version spark-streaming-kafka-0.8 spark-streaming-kafka-0.10
Broker Version 0.8.2.1 or higher 0.10.0 or higher
Api Maturity Deprecated Stable
Language Support Scala、Java、Python Scala、Java
Receiver DStream Yes No
Direct DStream Yes Yes
SSL / TLS Support No Yes
Offset Commit Api No Yes
Dynamic Topic Subscription No Yes

目前 CKafka 兼容 0.9及以上的版本，本次实践使用 0.10.2.1 版本的 Kafka 依赖。

此外，EMR 中的 Spark Streaming 也支持直接对接 CKafka，详见 SparkStreaming 对接 CKafka 服务。

一、步骤1：获取 CKafka 实例接入地址

1. 登录 CKafka 控制台。

2. 在左侧导航栏选择实例列表，单击实例的“ID”，进入实例基本信息页面。

3. 在实例的基本信息页面的接入方式模块，可获取实例的接入地址，接入地址是生产消费需要用到的 bootstrap-server。


二、步骤2：创建 Topic

1. 在实例基本信息页面，选择顶部Topic管理页签。

2. 在 Topic 管理页面，单击新建，创建一个名为 test 的 Topic，接下来将以该 Topic 为例介绍如何生产消费。


三、步骤3：准备云服务器环境

CentOS6.8 系统

package version
sbt 0.13.16
Hadoop 2.7.3
Spark 2.1.0
protobuf 2.5.0
ssh CentOS 默认安装
Java 1.8

具体安装步骤参见 配置环境。

四、步骤4：对接 CKafka

向 CKafka 中生产消息从 CKafka 消费消息

这里使用 0.10.2.1 版本的 Kafka 依赖。1. 在 build.sbt 添加依赖：name := "Producer Example"version := "1.0"scalaVersion := "2.11.8"libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.10.2.1"2. 配置 producer_example.scala： import java.util.Properties import org.apache.kafka.clients.producer._ object ProducerExample extends App { val props = new Properties() props.put("bootstrap.servers", "172.16.16.12:9092") //实例信息中的内网 IP 与端口 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer") props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer") val producer = new KafkaProducer[String, String](props) val TOPIC="test" //指定要生产的 Topic for(i<- 1 to 50){ val record = new ProducerRecord(TOPIC, "key", s"hello $i") //生产 key 是"key",value 是 hello i 的消息 producer.send(record) } val record = new ProducerRecord(TOPIC, "key", "the end "+new java.util.Date) producer.send(record) producer.close() //最后要断开 }更多有关 ProducerRecord 的用法请参见 ProducerRecord 文档。

五、DirectStream

1. 在 build.sbt 添加依赖：

name := "Consumer Example"version := "1.0"scalaVersion := "2.11.8"libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"libraryDependencies += "org.apache.spark" %% "spark-streaming" % "2.1.0"libraryDependencies += "org.apache.spark" %% "spark-streaming-kafka-0-10" % "2.1.0"

2. 配置 DirectStream_example.scala：

import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.serialization.StringDeserializerimport org.apache.kafka.common.TopicPartitionimport org.apache.spark.streaming.kafka010._import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistentimport org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribeimport org.apache.spark.streaming.kafka010.KafkaUtilsimport org.apache.spark.streaming.kafka010.OffsetRangeimport org.apache.spark.streaming.{Seconds, StreamingContext}import org.apache.spark.SparkConfimport org.apache.spark.SparkContextimport collection.JavaConversions._import Array._object Kafka { def main(args: Array[String]) { val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "172.16.16.12:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "spark_stream_test1", "auto.offset.reset" -> "earliest", "enable.auto.commit" -> "false" ) val sparkConf = new SparkConf() sparkConf.setMaster("local") sparkConf.setAppName("Kafka") val ssc = new StreamingContext(sparkConf, Seconds(5)) val topics = Array("spark_test") val offsets : Map[TopicPartition, Long] = Map() for (i <- 0 until 3){ val tp = new TopicPartition("spark_test", i) offsets.updated(tp , 0L) } val stream = KafkaUtils.createDirectStream[String, String]( ssc, PreferConsistent, Subscribe[String, String](topics, kafkaParams) ) println("directStream") stream.foreachRDD{ rdd=> //输出获得的消息 rdd.foreach{iter => val i = iter.value println(s"${i}") } //获得offset val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges rdd.foreachPartition { iter => val o: OffsetRange = offsetRanges(TaskContext.get.partitionId) println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}") } } // Start the computation ssc.start() ssc.awaitTermination() }}

六、RDD

1. 配置build.sbt（配置同上，单击查看）。

2. 配置RDD_example：

import org.apache.kafka.clients.consumer.ConsumerRecordimport org.apache.kafka.common.serialization.StringDeserializerimport org.apache.spark.streaming.kafka010._import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistentimport org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribeimport org.apache.spark.streaming.kafka010.KafkaUtilsimport org.apache.spark.streaming.kafka010.OffsetRangeimport org.apache.spark.streaming.{Seconds, StreamingContext}import org.apache.spark.SparkConfimport org.apache.spark.SparkContextimport collection.JavaConversions._import Array._object Kafka { def main(args: Array[String]) { val kafkaParams = Map[String, Object]( "bootstrap.servers" -> "172.16.16.12:9092", "key.deserializer" -> classOf[StringDeserializer], "value.deserializer" -> classOf[StringDeserializer], "group.id" -> "spark_stream", "auto.offset.reset" -> "earliest", "enable.auto.commit" -> (false: java.lang.Boolean) ) val sc = new SparkContext("local", "Kafka", new SparkConf()) val java_kafkaParams : java.util.Map[String, Object] = kafkaParams //按顺序向 partition 拉取相应 offset 范围的消息，如果拉取不到则阻塞直到超过等待时间或者新生产消息达到拉取的数量 val offsetRanges = Array[OffsetRange]( OffsetRange("spark_test", 0, 0, 5), OffsetRange("spark_test", 1, 0, 5), OffsetRange("spark_test", 2, 0, 5) ) val range = KafkaUtils.createRDD[String, String]( sc, java_kafkaParams, offsetRanges, PreferConsistent ) range.foreach(rdd=>println(rdd.value)) sc.stop() }}

更多 kafkaParams 用法参见 kafkaParams 文档。

七、安装 sbt

1. 在 sbt 页面 上下载 sbt 包。

2. 解压后在 sbt 的目录下创建一个 sbt_run.sh 脚本并增加可执行权限，脚本内容如下：

#!/bin/bashSBT_OPTS="-Xms512M -Xmx1536M -Xss1M -XX:+CMSClassUnloadingEnabled -XX:MaxPermSize=256M"java $SBT_OPTS -jar `dirname $0`/bin/sbt-launch.jar "$@"

chmod u+x ./sbt_run.sh

3. 执行以下命令。

./sbt-run.sh sbt-version

若能看到 sbt 版本说明可以正常运行。

八、安装 protobuf

1. 下载 protobuf 相应版本。

2. 解压后进入目录。

./configuremake && make install

需要预先安装 gcc-g++，执行中可能需要 root 权限。

3. 重新登录，在命令行中输入下述内容。

protoc –version

4. 若能看到 protobuf 版本说明可以正常运行。

九、安装 Hadoop

1. 访问 Hadoop 页面 下载所需要的版本。

2. 增加 Hadoop 用户。

useradd -m hadoop -s /bin/bash

十、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

3. 增加管理员权限。

visudo

十一、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

4. 在root ALL=(ALL) ALL下增加一行。hadoop ALL=(ALL) ALL保存退出。

5. 使用 Hadoop 进行操作。

su hadoop

6. SSH 无密码登录。

cd ~/.ssh/ # 若没有该目录，请先执行一次ssh localhostssh-keygen -t rsa # 会有提示，都按回车就可以cat id_rsa.pub >> authorized_keys # 加入授权chmod 600 ./authorized_keys # 修改文件权限

十二、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

7. 安装 Java。

sudo yum install java-1.8.0-openjdk java-1.8.0-openjdk-devel

8. 配置 ${JAVA_HOME}。

vim /etc/profile

在文末加上下述内容：

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.121-0.b13.el6_8.x86_64/jreexport PATH=$PATH:$JAVA_HOME

十三、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

根据安装情况修改对应路径。

9. 解压 Hadoop，进入目录。

./bin/hadoop version

若能显示版本信息说明能正常运行。

10. 配置单机伪分布式（可根据需要搭建不同形式的集群）。

vim /etc/profile

在文末加上下述内容：

export HADOOP_HOME=/usr/local/hadoopexport PATH=$HADOOP_HOME/bin:$PATH

十四、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

根据安装情况修改对应路径。

11. 修改 /etc/hadoop/core-site.xml。

<configuration> <property> <name>hadoop.tmp.dir</name> <value>file:/usr/local/hadoop/tmp</value> <description>Abase for other temporary directories.</description> </property> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property></configuration>

12. 修改 /etc/hadoop/hdfs-site.xml。

<configuration> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>file:/usr/local/hadoop/tmp/dfs/name</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>file:/usr/local/hadoop/tmp/dfs/data</value> </property></configuration>

13. 修改 /etc/hadoop/hadoop-env.sh 中的 JAVA_HOME 为 Java 的路径。

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.121-0.b13.el6_8.x86_64/jre

十五、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

14. 执行 NameNode 格式化。

./bin/hdfs namenode -format

显示Exiting with status 0则表示成功。

15. 启动 Hadoop。

./sbin/start-dfs.sh

成功启动会存在 NameNode 进程，DataNode 进程，SecondaryNameNode 进程。

十六、安装 Spark

访问 Spark 页面 下载所需要的版本。
因为之前安装了 Hadoop，所以选择使用 Pre-built with user-provided Apache Hadoop。

注意：本示例同样使用 Hadoop 用户进行操作。

1. 解压后进入目录。

2. 修改配置文件。

cp ./conf/spark-env.sh.template ./conf/spark-env.shvim ./conf/spark-env.sh

在第一行添加下述内容：

export SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)

十七、Running Environment

Operating System: Ubuntu 24.04.3 LTS / x86_64Runtime Version: GNU bash, version 5.2.21(1)-release (x86_64-pc-linux-gnu)

根据 Hadoop 安装情况修改路径。

3. 运行示例。

bin/run-example SparkPi

若成功安装可以看到程序输出 π 的近似值。

相关内容：

  1. 消息队列 CKafka 版 Flink 接入 CKafka Apache Flink 是一个可以处理流数据的实时处理框架，用于在无界和有界数据流上进行有状态的计算。Flink 能在所有常Apache Flink 擅长处理无界和有界数据集。Flink runtime 能够通过对时间和状态的精确控制处理无界数据流，也能够使用为固定大小数据集设计的算法和数据结构对有界数据集进行处理，并达到出色的性能。...
  2. 消息队列 CKafka 版 Schema Registry 接入 CKafka 无论是使用传统的 Avro API 自定义序列化类与反序列化类，还是使用 Twitter 的 Bijection 类库实现 Avro 的序列化与反序列化，两种方法有相同的缺点：在每条 Kafka 记录里都嵌入了 Schema，从而导致记录的大小成倍增加。但是不管怎样，在读取记录时仍然需要用到整个 Schema，所以要先找到 Schema。 CKafka 提供了数据共用一个 Schema 的方法：将 Schema 中的。...
  3. 弹性 MapReduce SparkStreaming 对接 Ckafka 服务 基于腾讯云的 EMR 服务可以以轻松结合腾讯云的 Ckafka 服务实现以下流式应用。...
  4. 数据开发治理平台 WeData DLC Spark Streaming Spark Streaming 作为 Spark 生态中专注于实时数据处理的核心组件，凭借其灵活的处理模式、与 Spark 核心引擎的深度融合以及丰富的数据源支持，成为实时大数据领域的重要工具。WeData 平台集成了 DLC Spark Streaming 节点，为用户提供了便捷的任务开发功能。本指南将详细讲解 DLC Spark Streaming 节点的配置方法。...
  5. 弹性 MapReduce Spark 集成 Kafka 说明 Spark 从2.3 起不再支持 Kafka0.8.2，EMR 现网版本已集成 Spark2.4.3及以上版本，需要集成 kafka 0.10.0及更高版本。...
  6. 消息队列 CKafka 版 Kafka Java 客户端问题 Kafka Java Producer 在2.3版本之前默认使用轮询分区策略进行消息发送，在2.4版本时候，开始提供可选的分区选择器，提供RoundRobinPartitioner 和 UniformStickyPartitioner 两种分区策略， 默认使用 UniformStickyPartitioner 轮询分区策略进行分区选择。 其中 RoundRobinPartitioner 存在未实现的方法 onNewB。...
  7. 消息队列 CKafka 版 Flume 接入 CKafka Apache Flume 是一个分布式、可靠、高可用的日志收集系统，支持各种各样的数据来源（如 HTTP、Log 文件、JMS、监听端口数据等），能将这些数据源的海量日志数据进行高效收集、聚合、移动，最后存储到指定存储系统中（如 Kafka、分布式文件系统、Solr 搜索服务器等）。...
  8. 消息队列 CKafka 版 Filebeat 接入 CKafka Beats 平台 集合了多种单一用途数据采集器。这些采集器安装后可用作轻量型代理，从成百上千或成千上万台机器向目标发送采集数据。  Beats 有多种采集器，可以以根据自身的需求下载对应的采集器。这篇以 Filebeat（轻量型日志采集器）为例，向你介绍 Filebeat 接入 CKafka 的操作方法，及接入后常。...
赞 (0)
上一篇 4天前
下一篇 4天前