助力工业物联网，工业大数据项目之数据采集

Maynor

发布于 2023-02-03 15:40:39

4970

发布于 2023-02-03 15:40:39

文章被收录于专栏：最新最全的大数据技术体系最新最全的大数据技术体系

文章目录

01：Sqoop命令回顾

目标：掌握Sqoop常用命令的使用

路径

step1：语法
step2：数据库参数
step3：导入参数
step4：导出参数
step5：其他参数

实施

语法

sqoop import | export \
--数据库连接参数
--HDFS或者Hive的连接参数
--配置参数

数据库参数

–connect jdbc:mysql://hostname:3306
–username
–password
–table
–columns
–where
-e/–query

导入参数

–delete-target-dir
–target-dir
–hcatalog-database
–hcatalog-table

导出参数

–export-dir
–hcatalog-database
–hcatalog-table

其他参数

-m

连接Oracle语法

--connect jdbc:oracle:thin:@OracleServer:OraclePort:OracleSID

测试采集Oracle数据

进入

docker exec -it sqoop bash

测试

sqoop import \
--connect jdbc:oracle:thin:@oracle.bigdata.cn:1521:helowin \
--username ciss \
--password 123456 \
--table CISS4.CISS_BASE_AREAS \
--target-dir /test/full_imp/ciss4.ciss_base_areas \
--fields-terminated-by "\t" \
-m 1

查看结果

小结

掌握Sqoop常用命令的使用

02：YARN资源调度及配置

目标：实现YARN的资源调度配置

实施

常用端口记住：排错

NameNode：8020,50070
ResourceManager：8032,8088
JobHistoryServer：19888
Master：7077,8080
HistoryServer：18080

YARN调度策略

FIFO：不用
- 单队列，队列内部FIFO，所有资源只给一个程序运行
Capacity：Apache
- 多队列，队列内部FIFO，资源分配给不同的队列，队列内部所有资源只给一个程序运行
Fair：CDH
- 多队列，队列内部共享资源，队列内部的资源可以给多个程序运行

YARN面试题

程序提交成功，但是不运行而且不报错，什么问题，怎么解决？
- 资源问题：APPMaster就没有启动
- 环境问题
  - NodeManager进程问题：进程存在，但不工作
  - 机器资源不足导致YARN或者HDFS服务停止：磁盘超过90%，所有服务不再工作
  - 解决：实现监控告警：80%，邮件告警
YARN中程序运行失败的原因遇到过哪些？
- 代码逻辑问题
- 资源问题：Container
  - Application / Driver：管理进程
  - MapTask和ReduceTask / Executor：执行进程
- 解决问题：配置进程给定更多的资源

问题1：程序已提交YARN，但是无法运行，报错：Application is added to the scheduler and is not activated. User’s AM resource limit exceeded.

yarn.scheduler.capacity.maximum-am-resource-percent=0.8

配置文件：${HADOOP_HOME}/etc/hadoop/capacity-scheduler.xml
属性功能：指定队列最大可使用的资源容量大小百分比，默认为0.2，指定越大，AM能使用的资源越多

问题2：程序提交，运行失败，报错：无法申请Container

yarn.scheduler.minimum-allocation-mb=512

配置文件：${HADOOP_HOME}/etc/hadoop/yarn-site.xml
属性功能：指定AM为每个Container申请的最小内存，默认为1G，申请不足1G，默认分配1G，值过大，会导致资源不足，程序失败，该值越小，能够运行的程序就越多

问题3：怎么提高YARN集群的并发度？

物理资源、YARN资源、Container资源、进程资源

YARN资源配置

yarn.nodemanager.resource.cpu-vcores=8
yarn.nodemanager.resource.memory-mb=8192

Container资源

yarn.scheduler.minimum-allocation-vcores=1
yarn.scheduler.maximum-allocation-vcores=32
yarn.scheduler.minimum-allocation-mb=1024
yarn.scheduler.maximum-allocation-mb=8192

MR Task资源

mapreduce.map.cpu.vcores=1
mapreduce.map.memory.mb=1024
mapreduce.reduce.cpu.vcores=1
mapreduce.reduce.memory.mb=1024

Spark Executor资源

--driver-memory  #分配给Driver的内存，默认分配1GB
--driver-cores   #分配给Driver运行的CPU核数，默认分配1核
--executor-memory #分配给每个Executor的内存数，默认为1G，所有集群模式都通用的选项
--executor-cores  #分配给每个Executor的核心数，YARN集合和Standalone集群通用的选项
--total-executor-cores NUM  #Standalone模式下用于指定所有Executor所用的总CPU核数
--num-executors NUM #YARN模式下用于指定Executor的个数，默认启动2个

实现：修改问题1中的配置属性

注意：修改完成，要重启YARN

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-jgRIa2kT-1673426702988)(Day2_数仓设计及数据采集.assets/image-20210822085238536.png)]

小结
- 实现YARN的资源调度配置

03：MR的Uber模式

目标：了解MR的Uber模式的配置及应用

实施

Spark为什么要比MR要快

MR慢

只有Map和Reduce阶段，每个阶段的结果都必须写入磁盘

如果要实现Map1 -> Map2 -> Reduce1 -> Reduce2
Mapreduce1：Map1
MapReduce2：Map2 -> Reduce1
Mapreduce3：Reduce2

MapReduce程序处理是进程级别：MapTask进程、ReduceTask进程

问题：MR程序运行在YARN上时，有一些轻量级的作业要频繁的申请资源再运行，性能比较差怎么办？

Uber模式

功能：Uber模式下，程序只申请一个AM Container：所有Map Task和Reduce Task，均在这个Container中顺序执行

默认不开启

配置：${HADOOP_HOME}/etc/hadoop/mapred-site.xml

mapreduce.job.ubertask.enable=true
#必须满足以下条件
mapreduce.job.ubertask.maxmaps=9
mapreduce.job.ubertask.maxreduces=1
mapreduce.job.ubertask.maxbytes=128M
yarn.app.mapreduce.am.resource.cpu-vcores=1
yarn.app.mapreduce.am.resource.mb=1536M

特点

Uber模式的进程为AM，所有资源的使用必须小于AM进程的资源
Uber模式条件不满足，不执行Uber模式
Uber模式，会禁用推测执行机制

小结

了解MR的Uber模式的配置及应用

04：Sqoop采集数据格式问题

目标：掌握Sqoop采集数据时的问题

路径

step1：现象
step2：问题
step3：原因
step4：解决

实施

现象

step1：查看Oracle中CISS_SERVICE_WORKORDER表的数据条数

select count(1) as cnt from CISS_SERVICE_WORKORDER;

step2：采集CISS_SERVICE_WORKORDER的数据到HDFS上

sqoop import –connect jdbc:oracle:thin:@oracle.bigdata.cn:1521:helowin –username ciss –password 123456 –table CISS4.CISS_SERVICE_WORKORDER –delete-target-dir –target-dir /test/full_imp/ciss4.ciss_service_workorder –fields-terminated-by “\001” -m 1

- step3：Hive中建表查看数据条数

- 进入Hive容器

  ```
docker exec -it hive bash
  ```

- 连接HiveServer

  ```
  beeline -u jdbc:hive2://hive.bigdata.cn:10000 -n root -p 123456
  ```

- 创建测试表

  ```sql
  create external table test_text(
  line string
  )
  location '/test/full_imp/ciss4.ciss_service_workorder';
  ```

- 统计行数

  ```
  select count(*) from test_text;
  ```

问题：Sqoop采集完成后导致HDFS数据与Oracle数据量不符

原因

sqoop以文本格式导入数据时，默认的换行符是特殊字符

Oracle中的数据列中如果出现了\n、\r、\t等特殊字符，就会被划分为多行

Oracle数据

id			name				age
001			zhang\nsan			18

Sqoop遇到特殊字段就作为一行

001			zhang
san			18

Hive

id			name				age
001			zhang 
san			18

解决

方案一：删除或者替换数据中的换行符
- –hive-drop-import-delims：删除换行符
- –hive-delims-replacement char：替换换行符
- 不建议使用：侵入了原始数据
方案二：使用特殊文件格式：AVRO格式

小结

掌握Sqoop采集数据时的问题

05：问题解决：Avro格式

目标：掌握使用Avro格式解决采集换行问题

路径

step1：常见格式介绍
step2：Avro格式特点
step3：Sqoop使用Avro格式
step4：使用测试

实施

常见格式介绍

类型	介绍
TextFile	Hive默认的文件格式，最简单的数据格式，便于查看和编辑，耗费存储空间，I/O性能较低
SequenceFile	含有键值对的二进制文件，优化磁盘利用率和I/O，并行操作数据，查询效率高，但存储空间消耗最大
AvroFile	特殊的二进制文件，设计的主要目标是为了满足schema evolution，Schema和数据保存在一起
OrcFile	列式存储，Schema存储在footer中，不支持schema evolution，高度压缩比并包含索引，查询速度非常快
ParquetFile	列式存储，与Orc类似，压缩比不如Orc，但是查询性能接近，支持的工具更多，通用性更强

SparkCore缺点：RDD【数据】：没有Schema
SparkSQL优点：DataFrame【数据 + Schema】
Schema：列的信息【名称、类型】

Avro格式特点

优点
- 二进制数据存储，性能好、效率高
- 使用JSON描述模式，支持场景更丰富
- Schema和数据统一存储，消息自描述
- 模式定义允许定义数据的排序
缺点
- 只支持Avro自己的序列化格式
- 少量列的读取性能比较差，压缩比较低
场景：基于行的大规模结构化数据写入、列的读取非常多或者Schema变更操作比较频繁的场景

Sqoop使用Avro格式

选项

--as-avrodatafile                                     Imports data to Avro datafiles

注意：如果使用了MR的Uber模式，必须在程序中加上以下参数避免类冲突问题

-Dmapreduce.job.user.classpath.first=true

使用测试

sqoop import \
-Dmapreduce.job.user.classpath.first=true \
--connect jdbc:oracle:thin:@oracle.bigdata.cn:1521:helowin \
--username ciss \
--password 123456 \
--table CISS4.CISS_SERVICE_WORKORDER \
--delete-target-dir \
--target-dir /test/full_imp/ciss4.ciss_service_workorder \
--as-avrodatafile \
--fields-terminated-by "\001" \
-m 1

Hive中建表

进入Hive容器

docker exec -it hive bash

连接HiveServer

beeline -u jdbc:hive2://hive.bigdata.cn:10000 -n root -p 123456

创建测试表

create external table test_avro(
line string
)
stored as avro
location '/test/full_imp/ciss4.ciss_service_workorder';

统计行数

select count(*) from test_avro;

小结

掌握如何使用Avro格式解决采集换行问题

06：Sqoop增量采集方案回顾

目标：回顾Sqoop增量采集方案

路径

step1：Append
step2：Lastmodified
step3：特殊方式

实施

Append

要求：必须有一列自增的值，按照自增的int值进行判断

特点：只能导入增加的数据，无法导入更新的数据

场景：数据只会发生新增，不会发生更新的场景

代码

sqoop import \
--connect jdbc:mysql://node3:3306/sqoopTest \
--username root \
--password 123456 \
--table tb_tohdfs \
--target-dir /sqoop/import/test02 \
--fields-terminated-by '\t' \
--check-column id \
--incremental append \
--last-value 0 \
-m 1

Lastmodified

要求：必须包含动态时间变化这一列，按照数据变化的时间进行判断

特点：既导入新增的数据也导入更新的数据

场景：一般无法满足要求，所以不用

代码

sqoop import \
--connect jdbc:mysql://node3:3306/sqoopTest \
--username root \
--password 123456 \
--table tb_lastmode \
--target-dir /sqoop/import/test03 \
--fields-terminated-by '\t' \
--incremental lastmodified \
--check-column lastmode \
--last-value '2021-06-06 16:09:32' \
-m 1

特殊方式

要求：每次运行的输出目录不能相同

特点：自己实现增量的数据过滤，可以实现新增和更新数据的采集

场景：一般用于自定义增量采集每天的分区数据到Hive

代码

sqoop  import \
--connect jdbc:mysql://node3:3306/db_order \
--username root \
--password-file file:///export/data/sqoop.passwd \
--query "select * from tb_order where substring(create_time,1,10) = '2021-09-14' or substring(update_time,1,10) = '2021-09-14' and \$CONDITIONS " \
--delete-target-dir \
--target-dir /nginx/logs/tb_order/daystr=2021-09-14 \
--fields-terminated-by '\t' \
-m 1

小结