Spark Streaming 是 Spark Core 的一个扩展,用于高吞吐且容错地处理持续性的数据,目前支持的外部输入有 Kafka、Flume、HDFS/S3、Kinesis、Twitter 和 TCP socket。
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 实例接入地址
- 登录 CKafka 控制台。
- 在左侧导航栏选择实例列表,单击实例的“ID”,进入实例基本信息页面。
- 在实例的基本信息页面的接入方式模块,可获取实例的接入地址,接入地址是生产消费需要用到的 bootstrap-server。
步骤2:创建 Topic
- 在实例基本信息页面,选择顶部Topic管理页签。
- 在 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
这里使用 0.10.2.1 版本的 Kafka 依赖。
- 在
build.sbt
添加依赖:name := "Producer Example" version := "1.0" scalaVersion := "2.11.8" libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.10.2.1"
- 配置
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 文档。
配置环境
安装 sbt
- 在 sbt 官网 上下载 sbt 包。
- 解压后在 sbt 的目录下创建一个 sbt_run.sh 脚本并增加可执行权限,脚本内容如下:
#!/bin/bash
SBT_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
- 执行以下命令。
./sbt-run.sh sbt-version
安装 protobuf
- 下载 protobuf 相应版本。
- 解压后进入目录。
./configure
make && make install - 重新登录,在命令行中输入下述内容。
protoc --version
- 若能看到 protobuf 版本说明可以正常运行。
安装 Hadoop
- 访问 Hadoop 官网 下载所需要的版本。
- 增加 Hadoop 用户。
useradd -m hadoop -s /bin/bash
- 增加管理员权限。
visudo
- 在
root ALL=(ALL) ALL
下增加一行。hadoop ALL=(ALL) ALL
保存退出。 - 使用 Hadoop 进行操作。
su hadoop
- SSH 无密码登录。
cd ~/.ssh/ # 若没有该目录,请先执行一次ssh localhost
ssh-keygen -t rsa # 会有提示,都按回车就可以
cat id_rsa.pub >> authorized_keys # 加入授权
chmod 600 ./authorized_keys # 修改文件权限 - 安装 Java。
sudo yum install java-1.8.0-openjdk java-1.8.0-openjdk-devel
- 配置 ${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/jre
export PATH=$PATH:$JAVA_HOME - 解压 Hadoop,进入目录。
./bin/hadoop version
- 配置单机伪分布式(可根据需要搭建不同形式的集群)。
vim /etc/profile
export HADOOP_HOME=/usr/local/hadoop
export PATH=$HADOOP_HOME/bin:$PATH - 修改
/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> - 修改
/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> - 修改
/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
- 执行 NameNode 格式化。
./bin/hdfs namenode -format
Exitting with status 0
则表示成功。 - 启动 Hadoop。
./sbin/start-dfs.sh
NameNode
进程,DataNode
进程,SecondaryNameNode
进程。
安装 Spark
访问 Spark 官网 下载所需要的版本。
因为之前安装了 Hadoop,所以选择使用 Pre-build with user-provided Apache Hadoop
。
说明:本示例同样使用
hadoop
用户进行操作。
- 解压进入目录。
- 修改配置文件。
cp ./conf/spark-env.sh.template ./conf/spark-env.sh
vim ./conf/spark-env.shexport SPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoop classpath)
- 运行示例。
bin/run-example SparkPi