Spark和Kafka环境安装

Posted 2021-04-03 苍兰诺

“Apache Spark是用于大规模数据处理的统一分析引擎。它提供Java，Scala，Python和R的高级API，以及支持常规执行图的优化引擎。它还支持一组丰富的更高级别的工具，包括星火SQL用于SQL和结构化数据的处理，MLlib机器学习，GraphX用于图形处理，以及结构化流的增量计算和流处理。”

。

01

—

Spark环境安装

spark-env.sh

#配置java的环境变量export JAVA_HOME=/kfly/install/jdk1.8.0_141#配置zk相关信息export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=node01:2181,node02:2181,node03:2181 -Dspark.deploy.zookeeper.dir=/spark"

注意

## 如果想运行spark-shell --master local[N] 读取HDFS上文件，则加上如下配置文件export HADOOP_CONF_DIR=/kfly/install/hadoop-2.6.0-cdh5.14.2/etc/hadoop# 这就是读取hdfs下，hdfs://node01:8020/words.txt 文件sc.textFile("/words.txt")

slaves

#指定spark集群的worker节点node02node03

分发到各个节点

# 1. 分发到各个节点scp -r /kfly/install/spark node02:/kfly/install# 2. 修改source /etc/profile环境export SPARK_HOME=/kfly/install/sparkexport PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin

02

—

Spark整合Hive

1、需要把hive安装目录下的配置文件hive-site.xml拷贝到每一个spark安装目录下对应的conf文件夹中 

2、需要一个连接mysql驱动的jar包拷贝到spark安装目录下对应的jars文件夹

3、可以使用spark-sql脚本 后期执行sql相关的任务

spark-sql \--master spark://node01:7077 \--executor-memory 1g \--total-executor-cores 4 \--conf spasrk.sql.warehouse.dir=hdfs://node01:8020/user/hive/warehouse

#!/bin/sh#定义sparksql提交脚本的头信息SUBMITINFO="spark-sql --master spark://node01:7077 --executor-memory 1g --total-executor-cores 4 --conf spark.sql.warehouse.dir=hdfs://node01:8020/user/hive/warehouse" #定义一个sql语句SQL="select * from kfly.psrson;" #执行sql语句 类似于 hive -e sql语句echo "$SUBMITINFO" echo "$SQL"$SUBMITINFO -e "$SQL"

02

—

Spark on Yarn

• http://spark.apache.org/docs/2.3.3/running-on-yarn.html

• spark-env.sh

## 注意这里不需要安装spark集群、只需要解压spark安装包到任意一台服务器#修改文件 spark-env.sh
#指定java的环境变量export JAVA_HOME=/kfly/install/jdk1.8.0_141#指定hadoop的配置文件目录export HADOOP_CONF_DIR=/kfly/install/hadoop-2.6.0-cdh5.14.2/etc/hadoop

spark-submit --class org.apache.spark.examples.SparkPi \--master yarn \--deploy-mode cluster \--driver-memory 1g \--executor-memory 1g \--executor-cores 1 \/kfly/install/spark-2.3.3-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.3.3.jar \10

• 如果运行出现错误，可能是虚拟内存不足，可以添加参数

• vim yarn-site.xml

<!--容器是否会执行物理内存限制默认为True--><property> <name>yarn.nodemanager.pmem-check-enabled</name> <value>false</value></property>
<!--容器是否会执行虚拟内存限制 默认为True--><property> <name>yarn.nodemanager.vmem-check-enabled</name> <value>false</value></property>

03

—

Kafka 环境搭建

• https://archive.apache.org/dist/kafka/1.0.1/kafka_2.11-1.0.1.tgz
  

• server.properties

#指定kafka对应的broker id ，唯一(eg：node01 0,node02 1,node03 2)broker.id=0#指定数据存放的目录log.dirs=/kfly/install/kafka/kafka-logs#指定zk地址zookeeper.connect=node01:2181,node02:2181,node03:2181#指定是否可以删除topic ,默认是false 表示不可以删除delete.topic.enable=true#指定broker主机名,node01 node02 node03host.name=node01

• 修改kafka环境变量

export KAFKA_HOME=/kfly/install/kafkaexport PATH=$PATH:$KAFKA_HOME/bin

• 分发到其他节点，然后运行

#!/bin/shcase $1 in "start"){for host in node01 node02 node03 do ssh $host "source /etc/profile; nohup /kfly/install/kafka/bin/kafka-server-start.sh /kfly/install/kafka/config/server.properties > /dev/null 2>&1 &"  echo "$host kafka is running..." done};;
"stop"){for host in node01 node02 node03 do ssh $host "source /etc/profile; nohup /kfly/install/kafka/bin/kafka-server-stop.sh >/dev/null 2>&1 &"  echo "$host kafka is stopping..." done};;esac

 kafka监控工具

• Kafka Manager

• https://github.com/yahoo/kafka-manager/releases

kafkaManager它是由雅虎开源的可以监控整个kafka集群相关信息的一个工具。（1）可以管理几个不同的集群（2）监控集群的状态(topics, brokers, 副本分布, 分区分布)（3）创建topic、修改topic相关配置

•  application.conf

#修改kafka-manager.zkhosts的值，指定kafka集群地址kafka-manager.zkhosts="node01:2181,node02:2181,node03:2181"

• 启动kafka-manager

• 启动zk集群，kafka集群，再使用root用户启动kafka-manager服务。

• bin/kafka-manager 默认的端口是9000，可通过 -Dhttp.port，指定端口

• -Dconfig.file=conf/application.conf指定配置文件

nohup bin/kafka-manager -Dconfig.file=conf/application.conf -Dhttp.port=8080 &

• KafkaOffsetMonitor

• https://github.com/quantifind/KafkaOffsetMonitor/tags
  

该监控是基于一个jar包的形式运行，部署较为方便。只有监控功能，使用起来也较为安全
(1)消费者组列表(2)查看topic的历史消费信息.(3)每个topic的所有parition列表(topic,pid,offset,logSize,lag,owner)(4)对consumer消费情况进行监控,并能列出每个consumer offset,滞后数据。

•  start_kafka_web.sh

#!/bin/shjava -cp KafkaOffsetMonitor-assembly-0.2.0.jar com.quantifind.kafka.offsetapp.OffsetGetterWeb --zk node01:2181,node02:2181,node03:2181 --port 8089 --refresh 10.seconds --retain 1.days# 启动脚本nohup sh start_kafka_web.sh &

• Kafka Eagle

• http://download.smartloli.org/

• 解压

• tar -zxvf kafka-eagle-bin-1.2.3.tar.gz -C /kkb/install

• 解压之后进入到kafka-eagle-bin-1.2.3目录中

• 得到kafka-eagle-web-1.2.3-bin.tar.gz

• 然后解压  tar -zxvf kafka-eagle-web-1.2.3-bin.tar.gz

• 重命名  mv kafka-eagle-web-1.2.3  kafka-eagle-web

• 修改配置文件

• 进入到conf目录

• 修改system-config.properties

# 填上你的kafka集群信息kafka.eagle.zk.cluster.alias=cluster1cluster1.zk.list=node01:2181,node02:2181,node03:2181
# kafka eagle页面访问端口kafka.eagle.webui.port=8048
# kafka sasl authenticatekafka.eagle.sasl.enable=falsekafka.eagle.sasl.protocol=SASL_PLAINTEXTkafka.eagle.sasl.mechanism=PLAINkafka.eagle.sasl.client=/kfly/install/kafka-eagle-bin-1.2.3/kafka-eagle-web/conf/kafka_client_jaas.conf
# 添加刚刚导入的ke数据库配置，我这里使用的是mysqlkafka.eagle.driver=com.mysql.jdbc.Driverkafka.eagle.url=jdbc:mysql://node02:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNullkafka.eagle.username=rootkafka.eagle.password=123456

• vi /etc/profile

 export KE_HOME=/kfly/install/kafka-eagle-web export PATH=$PATH:$KE_HOME/bin

• 启动kafka-eagle /访问

 sh bin/ke.sh start # 访问 `http://node01:8048/ke` # 用户名：admin password：123456