1 集群规划

Kafka是一个分布式的，高吞吐量，易于扩展的基于主题发布/订阅的消息系统。作为消息队列，kafka拥有高吞吐量，并且内置消息主题分区，备份，容错等特性，使得它非常适合应用在大规模的消息处理系统中。

Spark Streaming是对Spark Core的一个扩展，它能够实现对实时数据流的流式处理，并具有很好的可扩展性、高吞吐量和容错性。Spark Streaming支持从多种数据源读取数据，如：Kafka、Flume、Twitter、ZeroMQ、Kinesis以及TCP Sockets，并且可以提供一些高级API来表达复杂的处理算法，如：map、reduce、join和window等。最后，Spark Streaming支持将处理完的数据推送到文件系统、数据库中。

ZooKeeper是一个分布式的，开源的分布式应用程序协调服务，它包含一个简单的原语集，分布式应用程序可以基于它实现同步服务，配置维护和命名服务等。Zookeeper是hadoop的一个子项目，其发展历程无需赘述。在分布式应用中，由于工程师不能很好地使用锁机制，以及基于消息的协调机制不适合在某些应用中使用，因此需要有一种可靠的、可扩展的、分布式的、可配置的协调机制来统一系统的状态。Zookeeper的目的就在于此。

Redis是一个基于键值对的缓存与存储系统，通过提供多种键值数据类型来适应不同场景下的缓存与存储需求。由于数据存储在内存中，所以读取速度很快。

目前有7台节点，主机信息如下：

ip：192.168.10.121—127

用户名/密码：spark / spark

内存：128G

磁盘：36T

物理CPU：4

逻辑CPU：32

集群规划：

1.1 处理流程

Spark Streaming实时拉取kafka消息并解析，另一方面，定时器从oracle定时扫描活动。通过消息和活动的匹配，将符合条件的数据拼接成json串推送到kafka，与规则引擎对接。

2 环境搭建

2.1 系统环境

创建用户：

安装、运维及监控皆需使用 spark用户。

root 用户下执行如下命令进行添加 spark 用户。

useradd spark

passwd spark （输入两遍密码）

配置互信：

配置文件：

修改配置文件/etc/ssh/ssh_config

StrictHostKeyChecking no

修改配置文件/etc/ssh/sshd_config

PrintLastLog no

安装步骤：

步骤 1: 用ssh-key-gen在本地主机上创建公钥和密钥

ssh-keygen

步骤 2: 用ssh-copy-id 把公钥复制到远程主机上

ssh-copy-id spark@192.168.6.102

步骤 3: 直接登录远程主机

ssh 192.168.6.102

关闭防火墙：

root 用户下操作。

1.关闭防火墙服务

service iptables stop

2.设置防火墙服务器开机不自动启动

chkconfig iptables off

3.如果不关闭防火墙的话，需进行单独设置

集群内所有服务器之间需开放所有端口的 tcp 和 udp 端口。

修改主机名：

root用户下操作：

vi /etc/sysconfig/network

NETWORKING=yes

hostname=spark01

使用hostname spark01命令，下次登录生效。

修改主机名文件：

root 用户下进行操作。

修改/etc/hosts 文件

#第一列为 IP 地址，后面每列皆为 ip 地址的 hostname 别名

192.168.10.121 spark01

192.168.10.122 spark02

192.168.10.123 spark03

192.168.10.124 spark04

192.168.10.125 spark05

192.168.10.126 spark06

192.168.10.127 spark07

192.168.94.195 lncjmaster1

192.168.94.197 lncjslave1

192.168.94.198 lncjslave2

192.168.94.199 lncjslave3

192.168.94.200 lncjslave4

192.168.94.201 lncjslave5

192.168.94.202 lncjslave6

192.168.94.203 lncjslave7

2.2 软件环境

目录结构说明：

安装包存放目录：/opt/beh/core/

环境变量文件：/opt/beh/conf/beh_env

软件日志目录：/opt/beh/logs

安装jdk：

解压安装包，配置环境变量，所有节点都需要安装。

#JAVA 7u79

export JAVA_HOME=/opt/beh/core/jdk

安装scala：

解压安装包，配置环境变量, 所有节点都需要安装。

#SCALA 2.10.6

export SCALA_HOME=/opt/beh/core/scala

安装zookeeper：

zookeeper分别安装在spark01、spark02、spark03节点

修改配置文件/opt/beh/core/zookeeper/conf/zoo.cf

# The number of milliseconds of each tick

tickTime=2000

# The number of ticks that the initial

# synchronization phase can take

initLimit=10

# The number of ticks that can pass between

# sending a request and getting an acknowledgement

syncLimit=5

# the directory where the snapshot is stored.

# do not use /tmp for storage, /tmp here is just

# example sakes.

dataDir=/opt/beh/data/zookeeper

# the port at which the clients will connect

clientPort=2181

#

# Be sure to read the maintenance section of the

# administrator guide before turning on autopurge.

#

# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance

#

# The number of snapshots to retain in dataDir

autopurge.snapRetainCount=3

# Purge task interval in hours

# Set to "0" to disable auto purge feature

autopurge.purgeInterval=1

maxClientCnxns=0

server.1=spark01:2888:3888

server.2=spark02:2888:3888

server.3=spark03:2888:3888

修改/opt/beh/data/zookeeper/myid

##myid 文件的存放位置必须是 zoo.cfg 中参数。

dataDir=/opt/beh/data/zookeeper

##设置的目录。

##此文件中只有一个 id 是 zoo.cfg 中配置的对应关系。

##如下配置

server.1=spark01:2888:3888

server.2=spark02:2888:3888

server.3=spark03:2888:3888

##那么spark01 的 myid 文件必须是1， spark02 的myid 文件必须是2，spark03 的

myid 文件必须是3。

##myid 文件在不同服务器上配置。

实例：

[spark @spark01 ~]# cat /opt/beh/data/zookeeper/myid

1

[spark @ spark02 ~]# cat /opt/beh/data/zookeeper/myid

2

[spark @ spark03 ~]# cat /opt/beh/data/zookeeper/myid

3

命令：

a) 所有 zookeeper 节点分别执行：zkServer.sh start

b) 所有 zookeeper 节点上执行 jps 检测是否存在服务,显示为 QuorumPeerMain。

[spark@spark01 ~]$ jps

9360 Jps

8239 QuorumPeerMain

c) 验证：在命令行执行 zkCli.sh可以进入 zookeeper 服务。

并执行 ls / 显示查看存储信息，

[spark@spark01 ~]$ zkCli.sh

[zk: localhost:2181(CONNECTED) 0] ls /

[zookeeper]

再执行 quit 可退出交互。

d) 在每个 zookeeper 节点查看 zookeeper 角色 zkServer.sh status

##在所有 zookeeper 节点中随机生成一个 leader 角色，其他皆为 follower 角色。

安装hadoop：

spark依赖hdfs，具体安装参考先前文章

安装kafka：

kafka节点安装在spark04，spark05，spark06节点。

修改配置文件/opt/beh/core/kafka/config/server. properties，需要修改以下几个部分：

log.dirs=/disk01/kafka/kafka-logs

#zookeeper.connect=localhost:2181

zookeeper.connect=spark01:2181, spark:2181, spark:2181

# Timeout in ms for connecting to zookeeper

zookeeper.connection.timeout.ms=6000

delete.topic.enable=true

log.cleanup.policy=delete

log.retention.hours=24

启动命令：

nohup kafka-server-start.sh /opt/beh/core/kafka/config/server.properties > /dev/null &

安装redis：

安装参考

1.1 软件包

官方下载安装包redis-3.2.3，上传到/opt/beh/core目录并解压

[hadoop@hadoop013 core]$ tar zxf redis-3.2.3.tar.gz

[hadoop@hadoop013 core]$ mv redis-3.2.3 redis

[hadoop@hadoop013 core]$ cd redis

1.2 安装

1.2.1 编译

检查依赖

$ yum list gcc

编译

[hadoop@hadoop013 redis]$ make MALLOC=libc

……

Hint: It's a good idea to run 'make test' ;)

make[1]: Leaving directory `/opt/beh/core/redis/src'

出现如上提示说明编译成功。

如果是linux系统编译的时候执行make MALLOC=libc，如果是mac系统则执行make MALLOC=jemalloc。

1.2.2 配置

1.2.2.1 创建目录

创建以下目录

[hadoop@hadoop013 ~]$ mkdir /opt/beh/core/redis/{bin,conf,run}

[hadoop@hadoop013 ~]$ mkdir /opt/beh/data/redis

[hadoop@hadoop013 ~]$ mkdir /opt/beh/logs/redis

其他节点上同样创建数据与日志目录，即/opt/beh/data/redis与/opt/beh/logs/redis。

1.2.2.2 软链接

[hadoop@hadoop013 ~]$ cd /opt/beh/core/redis/bin

[hadoop@hadoop013 bin]$ ln -s /opt/beh/core/redis/src/redis-cli redis-cli

[hadoop@hadoop013 bin]$ ln -s /opt/beh/core/redis/src/redis-server redis-server

[hadoop@hadoop013 bin]$ ln -s /opt/beh/core/redis/src/redis-trib.rb redis-trib.rb

[hadoop@hadoop013 bin]$ ln -s /opt/beh/core/redis/src/redis-benchmark redis-benchmark

1.2.2.3 配置项

配置文件位置：/opt/beh/core/redis/conf/redis.conf

基本配置

bind 135.36.250.13

port 6379

daemonize yes

pidfile /opt/beh/core/redis/run/redis.pid

logfile /opt/beh/logs/redis/redis.log

dbfilename dump.rdb

dir /opt/beh/data/redis

#REDIS CLUSTER

cluster-enabled yes

cluster-config-file nodes.conf

cluster-node-timeout 15000

修改以上配置，redis便可以以集群方式正常启动使用。

配置内存与策略

maxmemory 10737418240

maxmemory-policy volatile-lru

考虑集群现有资源，设置节点最大使用内存为10G，策略为根据LRU算法生成的过期时间来删除key。

配置客户端连接数（默认10000）

maxclients 10000

1.2.3 启动服务

redis节点上执行

[hadoop@hadoop013 ~]$ redis-server /opt/beh/core/redis/conf/redis.conf

1.2.4 测试连接

1.2.4.1 查看进程

在节点分别执行ps可以看到：

[hadoop@hadoop013 ~]$ ps aux | grep redis

hadoop 9510 0.0 0.0 103324 888 pts/0 S+ 13:53 0:00 grep redis

hadoop 29595 0.0 0.0 129580 3648 ? Ssl 10:17 0:11 redis-server 135.36.250.13:6379 [cluster]

1.2.4.2 命令行

直接使用下面命令在任意一个节点上运行：

redis-cli -h 135.36.250.13 -p 6379 quit

显示OK表示连接成功后退出。

redis-cli -h 135.36.250.13 -p 6379 SHUTDOWN

安装spark：

spark01，spark02为master，其他节点为worker。

修改配置文件/opt/beh/core/spark/conf/slaves

[spark@spark04 conf]$ cat slaves

# A Spark Worker will be started on each of the machines listed below.

spark03

spark04

spark05

spark06

修改配置文件/opt/beh/core/spark/conf/spark-env.sh，具体参数根据实际情况设定。

#!/usr/bin/env bash

# This file is sourced when running various Spark programs.

# Copy it as spark-env.sh and edit that to configure Spark for your site.

# Options read when launching programs locally with

# ./bin/run-example or ./bin/spark-submit

# - HADOOP_CONF_DIR, to point Spark towards Hadoop configuration files

# - SPARK_LOCAL_IP, to set the IP address Spark binds to on this node

# - SPARK_PUBLIC_DNS, to set the public dns name of the driver program

# - SPARK_CLASSPATH, default classpath entries to append

# Options read by executors and drivers running inside the cluster

# - SPARK_LOCAL_IP, to set the IP address Spark binds to on this node

# - SPARK_PUBLIC_DNS, to set the public DNS name of the driver program

# - SPARK_CLASSPATH, default classpath entries to append

# - SPARK_LOCAL_DIRS, storage directories to use on this node for shuffle and RDD data

# - MESOS_NATIVE_JAVA_LIBRARY, to point to your libmesos.so if you use Mesos

# Options read in YARN client mode

# - HADOOP_CONF_DIR, to point Spark towards Hadoop configuration files

# - SPARK_EXECUTOR_INSTANCES, Number of workers to start (Default: 2)

# - SPARK_EXECUTOR_CORES, Number of cores for the workers (Default: 1).

# - SPARK_EXECUTOR_MEMORY, Memory per Worker (e.g. 1000M, 2G) (Default: 1G)

# - SPARK_DRIVER_MEMORY, Memory for Master (e.g. 1000M, 2G) (Default: 1G)

# - SPARK_YARN_APP_NAME, The name of your application (Default: Spark)

# - SPARK_YARN_QUEUE, The hadoop queue to use for allocation requests (Default: ‘default’)

# - SPARK_YARN_DIST_FILES, Comma separated list of files to be distributed with the job.

# - SPARK_YARN_DIST_ARCHIVES, Comma separated list of archives to be distributed with the job.

# Options for the daemons used in the standalone deploy mode

# - SPARK_MASTER_IP, to bind the master to a different IP address or hostname

# - SPARK_MASTER_PORT / SPARK_MASTER_WEBUI_PORT, to use non-default ports for the master

# - SPARK_MASTER_OPTS, to set config properties only for the master (e.g. "-Dx=y")

# - SPARK_WORKER_CORES, to set the number of cores to use on this machine

# - SPARK_WORKER_MEMORY, to set how much total memory workers have to give executors (e.g. 1000m, 2g)

# - SPARK_WORKER_PORT / SPARK_WORKER_WEBUI_PORT, to use non-default ports for the worker

# - SPARK_WORKER_INSTANCES, to set the number of worker processes per node

# - SPARK_WORKER_DIR, to set the working directory of worker processes

# - SPARK_WORKER_OPTS, to set config properties only for the worker (e.g. "-Dx=y")

# - SPARK_DAEMON_MEMORY, to allocate to the master, worker and history server themselves (default: 1g).

# - SPARK_HISTORY_OPTS, to set config properties only for the history server (e.g. "-Dx=y")

# - SPARK_SHUFFLE_OPTS, to set config properties only for the external shuffle service (e.g. "-Dx=y")

# - SPARK_DAEMON_JAVA_OPTS, to set config properties for all daemons (e.g. "-Dx=y")

# - SPARK_PUBLIC_DNS, to set the public dns name of the master or workers

# Generic options for the daemons used in the standalone deploy mode

# - SPARK_CONF_DIR Alternate conf dir. (Default: ${SPARK_HOME}/conf)

# - SPARK_LOG_DIR Where log files are stored. (Default: ${SPARK_HOME}/logs)

# - SPARK_PID_DIR Where the pid file is stored. (Default: /tmp)

# - SPARK_IDENT_STRING A string representing this instance of spark. (Default: $USER)

# - SPARK_NICENESS The scheduling priority for daemons. (Default: 0)

export JAVA_HOME=/opt/beh/core/jdk

export HADOOP_HOME=/opt/beh/core/hadoop

export HADOOP_CONF_DIR=/opt/beh/core/hadoop/etc/hadoop

export SPARK_HOME=/opt/beh/core/spark

export SPARK_CONF_DIR=/opt/beh/core/spark/conf

export SPARK_EXECUTOR_INSTANCES=2

export SPARK_EXECUTOR_CORES=1

export SPARK_EXECUTOR_MEMORY=4G

export SPARK_DRIVER_MEMORY=4G

export SPARK_YARN_APP_NAME=Spark

export SPARK_YARN_QUEUE=default

#export SPARK_YARN_DIST_FILES

#export SPARK_YARN_DIST_ARCHIVES

#export SPARK_WORKER_DIR=/opt/beh/core/spark/work

#export SPARK_LOG_DIR=/opt/beh/logs/spark

#export SPARK_PID_DIR=/opt/beh/tmp

#export SPARK_LOCAL_DIRS=/home/disk/tmp/spark/local

#export SPARK_CLASSPATH=/opt/beh/core/spark/lib/mysql-connector-java-5.1.31-bin.jar

#export SPARK_CLASSPATH=/opt/beh/core/spark/lib/mysql-connector-java-5.1.31-bin.jar:/opt/beh/core/spark/lib:/opt/beh/core/spark/lib/hadoop-hdfs-2.5.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/hadoop-common-2.5.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/hadoop-core-2.5.0-mr1-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-bagel_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-catalyst_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-core_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-graphx_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-hive_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-mllib_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-repl_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-sql_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-streaming_2.10-1.1.0-cdh5.2.0.jar:/opt/beh/core/spark/lib/spark-yarn_2.10-1.1.0-cdh5.2.0.jar

export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=zk01,zk02,zk03:2181 -Dspark.deploy.zookeeper.dir=/opt/beh/logs/zookeeper/spark"

export LD_LIBRARY_PATH=/opt/beh/core/hadoop/lib/native

export SPARK_LIBRARY_PATH=${SPARK_HOME}/lib:$LD_LIBRARY_PATH

修改配置文件/opt/beh/core/spark/conf/spark-defaults.conf，具体参数根据实际情况设定。

# Default system properties included when running spark-submit.

# This is useful for setting default environmental settings.

# Example:

# spark.master spark://master:7077

# spark.eventLog.enabled true

# spark.eventLog.dir hdfs://namenode:8021/directory

# spark.serializer org.apache.spark.serializer.KryoSerializer

# spark.driver.memory 5g

# spark.executor.extraJavaOptions -XX:+PrintGCDetails -Dkey=value -Dnumbers="one two three"

spark.master spark://spark01:7077,spark02:7077

spark.eventLog.enabled false

#spark.eventLog.dir hdfs://beh/SparkeventLog

#spark.eventLog.compress false

spark.serializer org.apache.spark.serializer.KryoSerializer

spark.driver.memory 4g

spark.executor.memory 4g

spark.executor.cores 1

spark.executor.extraClassPath /opt/beh/core/spark/lib/mysql-connector-java-5.1.31-bin.jar

spark.driver.extraClassPath /opt/beh/core/spark/lib/mysql-connector-java-5.1.31-bin.jar:/opt/beh/core/hive/lib/AuthHook.jar

spark.app.id Spark_App

spark.executor.extraJavaOptions -XX:+PrintGCDetails -Dkey=value -Dnumbers="one two three"

spark.local.dir /opt/beh/data/spark/local

###Port for your application's dashboard, which shows memory and workload data

spark.ui.port 4040

spark.broadcast.compress true

spark.rdd.compress false

spark.io.compression.codec snappy

spark.io.compression.snappy.blockSize 32768

##Default number of tasks to use across the cluster for distributed shuffle operations (groupByKey, reduceByKey, etc) when not set by user.

spark.default.parallelism 4

spark.cores.max 4

启动服务命令：

在spark01执行 start-master.sh，start-slaves.sh;

查看master进程：

[spark@spark01 ~]$ jps

38397 DFSZKFailoverController

4454 ResourceManager

4829 Master

4100 NameNode

3865 QuorumPeerMain

4315 JournalNode

42675 Jps

在spark02 执行 start-master.sh；

在其他节点查看worker进程：

[spark@spark03 ~]$ jps

4048 JournalNode

3844 QuorumPeerMain

3959 DataNode

4170 NodeManager

4364 Worker

56466 Jps

停止服务,在对应的节点执行以下命令：

stop-master.sh

stop-slaves.sh