黑马一站制造数仓实战2

问题
DG连接问题
原理:JDBC:用Java代码连接数据库
Hive/SparkSQL:端口有区别
可以为同一个端口,只要不在同一台机器
项目:一台机器

HiveServer:10000

hiveserver.port = 10000

SparkSQL:10001

start-thriftserver.sh --hiveserver.prot = 10001

MySQL:hostname、port、username、password
Oracle:hostname、port、username、password、sid

CS模式设计问题
Thrift启动问题
CS模式:客户端服务端模式
Client:客户端
Hive:Beeline、Hue
SparkSQL
Server:服务端
Hive:Hiveserver2【负责解析SQL语句】
HiveServer作为Metastore的客户端
MetaStore作为HiveServer的服务端
SparkSQL:ThriftServer【负责解析SQL语句转换为SparkCore程序】

启动ThriftServer或者HiveServer

docker start hadoop
docker start hive
docker start spark

问题:思路
现象:异常
Python:error:xxxxxx
Java:throw Exception:xxxxxxxxx
进程没有明显报错:找日志文件
日志文件:logs
查看日志:tail -100f logs/xxxxxxxx.log
分析错误
ArrayoutofIndex
NullException
ClassNotFound

数据仓库设计
建模:维度建模:【事实表、维度表】
分层:ODS、DW【DWD、DWM、DWS】、APP
本次项目中数仓的分层
ODS、DWD、DWB、DWS、ST、DM

数仓设计回顾
目标:了解数据仓库设计的核心知识点
路径
step1:分层
step2:建模
实施
分层
什么是分层?
本质:规范化数据的处理流程
实现:每一层在Hive中就是一个数据库
为什么要分层?
清晰数据结构:每一个数据分层都有它的作用域,这样我们在使用表的时候能更方便地定位和解。
数据血缘追踪:简单来讲可以这样理解,我们最终给业务诚信的是一能直接使用的一张业务表,但是它的来源有很多,如果有一张来源表出问题了,我们希望能够快速准确地定位到问题,并清楚它的危害范围。
减少重复开发:规范数据分层,开发一些通用的中间层数据,能够减少极大的重复计算。
把复杂问题简单化:一个复杂的任务分解成多个步骤来完成,每一层只处理单一的步骤,比较简单和容易理解。
屏蔽原始数据的异常对业务的影响:不必改一次业务就需要重新接入数据
怎么分层?
ODS:原始数据层/操作数据层,最接近与原始数据的层次,数据基本与原始数据
保持一致
DW:数据仓库层,实现数据的处理转换
DWD:实现ETL
DWM:轻度聚合
DWS:最终聚合
ADS/APP/DA:数据应用层
建模
什么是建模?
本质:决定了数据存储的方式,表的设计
为什么要建模?
大数据系统需要数据模型方法来帮助更好地组织和存储数据,以便在性能、成本、效率和质量之间取得最佳平衡。
性能:良好的数据模型能帮助我们快速查询所需要的数据,减少数据的I/O吞吐
成本:良好的数据模型能极大地减少不必要的数据冗余,也能实现计算结果复用,极大地降低大数据系统中的存储和计算成本
效率:良好的数据模型能极大地改善用户使用数据的体验,提高使用数据的效率
质量:良好的数据模型能改善数据统计口径的不一致性,减少数据计算错误的可能性
有哪些建模方法?
ER模型:从全企业的高度设计一个 3NF 【三范式】模型,用实体关系模型描述企业业务,满足业务需求的存储
维度模型:从分析决策的需求出发构建模型,为分析需求服务,重点关注用户如何更快速的完成需求分析,具有较好的大规模复杂查询的响应性能
Data Vault:ER 模型的衍生,基于主题概念将企业数据进行结构化组织,并引入了更进一步的范式处理来优化模型,以应对源系统变更的扩展性
Anchor:一个高度可扩展的模型,核心思想是所有的扩展知识添加而不是修改,因此将模型规范到 6NF,基本变成了 k-v 结构化模型
怎么构建维度模型步骤?
a.选择业务过程:你要做什么?
b.声明粒度:你的分析基于什么样的颗粒度?
c.确认环境的维度:你的整体有哪些维度?
d.确认用于度量的事实:你要基于这些维度构建哪些指标?
具体的实施流程是什么?
a.需求调研:业务调研和数据调研
业务调研:明确分析整个业务实现的过程
数据调研:数据的内容是什么
b.划分主题域:面向业务将业务划分主题
构建哪些主题域以及每个主题域中有哪些主题
服务域:工单主题、回访主题、物料主题
c.构建维度总线矩阵:明确每个业务主题对应的维度关系

d.明确指标统计:明确所有原生指标与衍生指标
工单主题:安装工单个数、维修工单个数……
回访主题:用户满意个数、不满意个数、服务态度不满意个数、技术能力不满意个数
e.定义事实与维度规范
分层规范
开发规范
……
f.代码开发
事实表
表的分类
事务事实表:原始的事务事实的数据表,原始业务数据表
周期快照事实表:周期性对事务事实进行聚合的结果
累计快照事实表:随着时间的变化,事实是不定的,不断完善的过程
无事实事实表:特殊的事实表,里面没有事实,是多个维度的组合,用于求事实的差值
值的分类
可累加事实:在任何维度下指标的值都可以进行累加
半可累加事实:在一定维度下指标的值都可以进行累加
不可累加事实:在任何维度下指标的值都不可以进行累加
维度表
维度设计模型
雪花模型:维度表拥有子维度表,部分维度表关联在维度表中,间接的关联事实表
星型模型/星座模型:维度表没有子维度,直接关联在事实表上,星座模型中有多个事实
上卷与下钻
上卷:从小维度到一个大的维度,颗粒度从细到粗
下钻:从大维度到一个小的维度,颗粒度从粗到细
拉链表
功能:解决事实中渐变维度发生变化的问题,通过时间来标记维度的每一种状态,存储所有状态
实现
step1:先采集所有增量数据到更新表中
step2:将更新表的数据与老的拉链表的数据进行合并写入一张临时表
step3:将临时表的结果覆盖到拉链表中

掌握油站分析项目中的分层整体设计

ODS:原始数据层:最接近于原始数据的层次,直接采集写入层次:原始事务事实表
DWD:明细数据层:对ODS层的数据根据业务需求实现ETL以后的结果:ETL以后事务事实表
DWB:基础数据层:类似于以前讲解的DWM,轻度聚合
关联:将主题事实的表进行关联,所有与这个主题相关的字段合并到一张表
聚合:基于主题的事务事实构建基础指标
主题事务事实表
ST:数据应用层:类似于以前讲解的APP,存储每个主题基于维度分析聚合的结果:周期快照
事实表
供数据分析的报表
DM:数据集市:按照不同部门的数据需求,将暂时没有实际主题需求的数据存储
做部门数据归档,方便以后新的业务需求的迭代开发
DWS:维度数据层:类似于以前讲解的DIM:存储维度数据表
数据仓库设计方案
从上到下:在线教育:先明确需求和主题,然后基于主题的需求采集数据,处理数据
场景:数据应用比较少,需求比较简单

掌握油站分析的每层的具体功能
实施
ODS
数据内容:存储所有原始业务数据,基本与Oracle数据库中的业务数据保持一致
数据来源:使用Sqoop从Oracle中同步采集
存储设计:Hive分区表,avro文件格式存储,保留3个月
DWD
数据内容:存储所有业务数据的明细数据
数据来源:对ODS层的数据进行ETL扁平化处理得到
存储设计:Hive分区表,orc文件格式存储,保留所有数据
DWB
数据内容:存储所有事实与维度的基本关联、基本事实指标等数据
数据来源:对DWD层的数据进行清洗过滤、轻度聚合以后的数据
存储设计:Hive分区表,orc文件格式存储,保留所有数据
ST
数据内容:存储所有报表分析的事实数据
数据来源:基于DWB和DWS层,通过对不同维度的统计聚合得到所有报表事实的指标
DM
数据内容:存储不同部门所需要的不同主题的数据
数据来源:对DW层的数据进行聚合统计按照不同部门划分
DWS
数据内容:存储所有业务的维度数据:日期、地区、油站、呼叫中心、仓库等维度表
数据来源:对DWD的明细数据中抽取维度数据
存储设计:Hive普通表,orc文件 + Snappy压缩
特点:数量小、很少发生变化、全量采集

全量表与增量表数据采集需求
实施
全量表
所有维度数据表
场景:不会经常发生变化的数据表,例如维度数据表等
数据表:组织机构信息、地区信息、服务商信息、数据字典等
表名:参考文件《full_import_tables.txt》
增量表
所有事务事实的数据表
场景:经常发生变化的数据表,例如业务数据、用户行为数据等
数据表:工单数据信息、呼叫中心信息、物料仓储信息、报销费用信息等
表名:参考文件《incr_import_tables.txt》

Sqoop命令

连接Oracle语法

--connect jdbc:oracle:thin:@OracleServer:OraclePort:OracleSID 1
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

YARN常用端口

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

程序提交成功,但是不运行而且不报错,什么问题,怎么解决?
资源问题:APPMaster就没有启动
环境问题
NodeManager进程问题:进程存在,但不工作
机器资源不足导致YARN或者HDFS服务停止:磁盘超过90%,所有服务不再工作
解决:实现监控告警:80%,邮件告警

YARN中程序运行失败的原因遇到过哪些?
代码逻辑问题
资源问题:Container
Application / Driver:管理进程
MapTask和ReduceTask / Executor:执行进程
解决问题:配置进程给定更多的资源

程序已提交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能使用的资源越多

程序提交,运行失败,报错:无法申请Container?

yarn.scheduler.minimum-allocation-mb=512

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

怎么提高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个

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模式,会禁用推测执行机制

Sqoop采集数据格式问题

现象
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

创建测试表

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数据

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

Hive

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

常见格式介绍

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

Sqoop增量采集方案

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

实现自动化脚本开发

脚本目标:实现自动化将多张Oracle中的数据表全量或者增量采集同步到HDFS中
实现流程
a. 获取表名
b. 构建Sqoop命令
c. 执行Sqoop命令
d. 验证结果
脚本选型
Shell:Linux原生Shell脚本,命令功能全面丰富,主要用于实现自动化Linux指令,适合
于Linux中简单的自动化任务开发
Python:多平台可移植兼容脚本,自身库功能强大,主要用于爬虫、数据科学分析计算
等,适合于复杂逻辑的处理计算场景
场景:一般100行以内的代码建议用Shell,超过100行的代码建议用Python
采集脚本选用:Shell
单个测试
创建一个文件,存放要采集的表的名称

#创建测试目录
mkdir -p /opt/datas/shell
cd /opt/datas/shell/
#创建存放表名的文件
vim test_full_table.txt

ciss4.ciss_base_areas
ciss4.ciss_base_baseinfo
ciss4.ciss_base_csp
ciss4.ciss_base_customer
ciss4.ciss_base_device

创建脚本

vim test_full_import_table.sh

构建采集的Sqoop命令

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

 封装脚本

#!/bin/bash
#export path
source /etc/profile
#export the tbname files
TB_NAME=/opt/datas/shell/test_full_table.txt
#export the import opt
IMP_OPT="sqoop import -
Dmapreduce.job.user.classpath.first=true"
#export the jdbc opt
JDBC_OPT="--connect
jdbc:oracle:thin:@oracle.bigdata.cn:1521:helowin --username
ciss --password 123456"
#read tbname and exec sqoop
while read tbname
do
${IMP_OPT} ${JDBC_OPT} --table ${tbname^^} --delete-target-dir --target-dir /test/full_imp/${tbname^^} --as-avrodatafile --fields-terminated-by "\001" -m 1
done < ${TB_NAME}

添加执行权限

chmod u+x test_full_import_table.sh

测试执行

sh -x test_full_import_table.sh

检查结果

实现全量采集脚本的运行
实施
全量目标:将所有需要将实现全量采集的表进行全量采集存储到HDFS上
Oracle表:组织机构信息、地区信息、服务商信息、数据字典等
HDFS路径

/data/dw/ods/one_make/full_imp/表名/日期

增量目标:将所有需要将实现全量采集的表进行增量采集存储到HDFS上
工单数据信息、呼叫中心信息、物料仓储信息、报销费用信息等
HDFS路径

/data/dw/ods/one_make/incr_imp/表名/日期

导入全量表

#!/usr/bin/env bash
# /bin/bash
biz_date=20210101
biz_fmt_date=2021-01-01
dw_parent_dir=/data/dw/ods/one_make/full_imp
workhome=/opt/sqoop/one_make
full_imp_tables=${workhome}/full_import_tables.txt
mkdir ${workhome}/logorcl_srv=oracle.bigdata.cn
orcl_port=1521
orcl_sid=helowin
orcl_user=ciss
orcl_pwd=123456sqoop_import_params="sqoop import -Dmapreduce.job.user.classpath.first=true --outdir ${workhome}/java_code --as-avrodatafile"
sqoop_jdbc_params="--connect jdbc:oracle:thin:@${orcl_srv}:${orcl_port}:${orcl_sid} --username ${orcl_user} --password ${orcl_pwd}"# load hadoop/sqoop env
source /etc/profilewhile read p; do# parallel execution import${sqoop_import_params} ${sqoop_jdbc_params} --target-dir ${dw_parent_dir}/${p}/${biz_date} --table ${p^^} -m 1 &cur_time=`date "+%F %T"`echo "${cur_time}: ${sqoop_import_params} ${sqoop_jdbc_params} --target-dir ${dw_parent_dir}/${p}/${biz_date} --table ${p} -m 1 &" >> ${workhome}/log/${biz_fmt_date}_full_imp.logsleep 30
done < ${full_imp_tables}

运行脚本
全量采集

cd /opt/sqoop/one_make
sh -x full_import_tables.sh

脚本中特殊的一些参数
--outdir:Sqoop解析出来的MR的Java程序等输出文件输出的文件

导入增量表

#!/usr/bin/env bash
# 编写SHELL脚本的时候要特别小心,特别是编写SQL的条件,如果中间加了空格,就会导致命令执行失败
# /bin/bash
biz_date=20210101
biz_fmt_date=2021-01-01
dw_parent_dir=/data/dw/ods/one_make/incr_imp
workhome=/opt/sqoop/one_make
incr_imp_tables=${workhome}/incr_import_tables.txtorcl_srv=oracle.bigdata.cn
orcl_port=1521
orcl_sid=helowin
orcl_user=ciss
orcl_pwd=123456mkdir ${workhome}/logsqoop_condition_params="--where \"'${biz_fmt_date}'=to_char(CREATE_TIME,'yyyy-mm-dd')\""
sqoop_import_params="sqoop import -Dmapreduce.job.user.classpath.first=true --outdir ${workhome}/java_code --as-avrodatafile"
sqoop_jdbc_params="--connect jdbc:oracle:thin:@${orcl_srv}:${orcl_port}:${orcl_sid} --username ${orcl_user} --password ${orcl_pwd}"# load hadoop/sqoop env
source /etc/profilewhile read p; do# clean old directory in HDFShdfs dfs -rm -r ${dw_parent_dir}/${p}/${biz_date}# parallel execution import${sqoop_import_params} ${sqoop_jdbc_params} --target-dir ${dw_parent_dir}/${p}/${biz_date} --table ${p^^} ${sqoop_condition_params} -m 1 &cur_time=`date "+%F %T"`echo "${cur_time}: ${sqoop_import_params} ${sqoop_jdbc_params} --target-dir ${dw_parent_dir}/${p}/${biz_date} --table ${p} ${sqoop_condition_params} -m 1 &" >> ${workhome}/log/${biz_fmt_date}_incr_imp.logsleep 30done < ${incr_imp_tables}

增量采集

cd /opt/sqoop/one_make
sh -x incr_import_tables.sh

特殊问题
因oracle表特殊字段类型,导致sqoop导数据任务失败
oracle字段类型为: clob或date等特殊类型
解决方案:在sqoop命令中添加参数,指定特殊类型字段列(SERIAL_NUM)的数据类型为string
—map-column-java SERIAL_NUM=String
查看结果
/data/dw/ods/one_make/full_imp:44张表
/data/dw/ods/one_make/incr_imp:57张表

Schema备份及上传
目标:了解如何实现采集数据备份
实施
需求:将每张表的Schema进行上传到HDFS上,归档并且备份
Avro文件本地存储

workhome=/opt/sqoop/one_make
--outdir ${workhome}/java_code

Avro文件HDFS存储

hdfs_schema_dir=/data/dw/ods/one_make/avsc
hdfs dfs -put ${workhome}/java_code/*.avsc ${hdfs_schema_dir}

Avro文件本地打包

local_schema_backup_filename=schema_${biz_date}.tar.gz
tar -czf ${local_schema_backup_filename} ./java_code/*.avsc

Avro文件HDFS备份

hdfs_schema_backup_filename=${hdfs_schema_dir}/avro_schema_${biz_date
}.tar.gz
hdfs dfs -put ${local_schema_backup_filename}
${hdfs_schema_backup_filename}

上传avro schema sh代码

#!/usr/bin/env bash
# 上传
# /bin/bash
workhome=/opt/sqoop/one_make
hdfs_schema_dir=/data/dw/ods/one_make/avsc
biz_date=20210101
biz_fmt_date=2021-01-01
local_schema_backup_filename=schema_${biz_date}.tar.gz
hdfs_schema_backup_filename=${hdfs_schema_dir}/avro_schema_${biz_date}.tar.gz
log_file=${workhome}/log/upload_avro_schema_${biz_fmt_date}.log# 打印日志
log() {cur_time=`date "+%F %T"`echo "${cur_time} $*" >> ${log_file}
}source /etc/profile
cd ${workhome}#  hadoop fs [generic options] [-test -[defsz] <path>]
# -test -[defsz] <path> :
#   Answer various questions about <path>, with result via exit status.
#     -d  return 0 if <path> is a directory.
#     -e  return 0 if <path> exists.
#     -f  return 0 if <path> is a file.
#     -s  return 0 if file <path> is greater than zero bytes in size.
#     -z  return 0 if file <path> is zero bytes in size, else return 1.log "Check if the HDFS Avro schema directory ${hdfs_schema_dir}..."
hdfs dfs -test -e ${hdfs_schema_dir} > /dev/nullif [ $? != 0 ]; thenlog "Path: ${hdfs_schema_dir} is not exists. Create a new one."log "hdfs dfs -mkdir -p ${hdfs_schema_dir}"hdfs dfs -mkdir -p ${hdfs_schema_dir}
filog "Check if the file ${hdfs_schema_dir}/CISS4_CISS_BASE_AREAS.avsc has uploaded to the HFDS..."
hdfs dfs -test -e ${hdfs_schema_dir}/CISS4_CISS_BASE_AREAS.avsc.avsc > /dev/null
if [ $? != 0 ]; thenlog "Upload all the .avsc schema file."log "hdfs dfs -put ${workhome}/java_code/*.avsc ${hdfs_schema_dir}"hdfs dfs -put ${workhome}/java_code/*.avsc ${hdfs_schema_dir}
fi# backup
log "Check if the backup tar.gz file has generated in the local server..." 
if [ ! -e ${local_schema_backup_filename} ]; thenlog "package and compress the schema files"log "tar -czf ${local_schema_backup_filename} ./java_code/*.avsc"tar -czf ${local_schema_backup_filename} ./java_code/*.avsc
filog "Check if the backup tar.gz file has upload to the HDFS..."
hdfs dfs -test -e ${hdfs_schema_backup_filename} > /dev/null
if [ $? != 0 ]; thenlog "upload the schema package file to HDFS"log "hdfs dfs -put ${local_schema_backup_filename} ${hdfs_schema_backup_filename}"hdfs dfs -put ${local_schema_backup_filename} ${hdfs_schema_backup_filename}
fi

运行测试

cd /opt/sqoop/one_make/
./upload_avro_schema.sh

验证结果

/data/dw/ods/one_make/avsc/
*.avsc
schema_20210101.tar.gz

使用Python脚本如何实现

导入全量表

#!/usr/bin/env python
# @Time : 2021/7/14 15:34
# @desc :
__coding__ = "utf-8"
__author__ = "itcast"import os
import subprocess
import datetime
import time
import loggingbiz_date = '20210101'
biz_fmt_date = '2021-01-01'
dw_parent_dir = '/data/dw/ods/one_make/full_imp'
workhome = '/opt/sqoop/one_make'
full_imp_tables = workhome + '/full_import_tables.txt'
if os.path.exists(workhome + '/log'):os.system('make ' + workhome + '/log')orcl_srv = 'oracle.bigdata.cn'
orcl_port = '1521'
orcl_sid = 'helowin'
orcl_user = 'ciss'
orcl_pwd = '123456'sqoop_import_params = 'sqoop import -Dmapreduce.job.user.classpath.first=true --outdir %s/java_code --as-avrodatafile' % workhome
sqoop_jdbc_params = '--connect jdbc:oracle:thin:@%s:%s:%s --username %s --password %s' % (orcl_srv, orcl_port, orcl_sid, orcl_user, orcl_pwd)# load hadoop/sqoop env
subprocess.call("source /etc/profile", shell=True)
print('executing...')
# read file
fr = open(full_imp_tables)
for line in fr.readlines():tblName = line.rstrip('\n')# parallel execution import# ${sqoop_import_params} ${sqoop_jdbc_params} --target-dir ${dw_parent_dir}/${p}/${biz_date} --table ${p^^} -m 1 &# sqoopImportCommand = f''' {sqoop_import_params} {sqoop_jdbc_params} --target-dir {dw_parent_dir}/{tblName}/{biz_date} --table {tblName.upper()} -m 1 &'''sqoopImportCommand = '''%s %s --target-dir %s/%s/%s --table %s -m 1 &''' % (sqoop_import_params, sqoop_jdbc_params, dw_parent_dir, tblName, biz_date, tblName.upper())# parallel execution importsubprocess.call(sqoopImportCommand, shell=True)# cur_time=`date "+%F %T"`# cur_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')logging.basicConfig(level=logging.INFO,  # 控制台打印的日志级别filename='%s/log/%s_full_imp.log' % (workhome, biz_fmt_date),# 模式,有w和a,w就是写模式,每次都会重新写日志,覆盖之前的日志; a是追加模式,默认如果不写的话,就是追加模式filemode='a',# 日志格式format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s')# logging.info(cur_time + ' : ' + sqoopImportCommand)logging.info(sqoopImportCommand)time.sleep(15)

导入增量表

#!/usr/bin/env python
# @Time : 2021/7/20 15:19
# @desc :
__coding__ = "utf-8"
__author__ = "itcast"import os
import subprocess
import datetime
import time
import loggingbiz_date = '20210101'
biz_fmt_date = '2021-01-01'
dw_parent_dir = '/data/dw/ods/one_make/incr_imp'
workhome = '/opt/sqoop/one_make'
incr_imp_tables = workhome + '/incr_import_tables.txt'
if os.path.exists(workhome + '/log'):os.system('make ' + workhome + '/log')orcl_srv = 'oracle.bigdata.cn'
orcl_port = '1521'
orcl_sid = 'helowin'
orcl_user = 'ciss'
orcl_pwd = '123456'sqoop_import_params = 'sqoop import -Dmapreduce.job.user.classpath.first=true --outdir %s/java_code --as-avrodatafile' % workhome
sqoop_jdbc_params = '--connect jdbc:oracle:thin:@%s:%s:%s --username %s --password %s' % (orcl_srv, orcl_port, orcl_sid, orcl_user, orcl_pwd)# load hadoop/sqoop env
subprocess.call("source /etc/profile", shell=True)
print('executing...')
# read file
fr = open(incr_imp_tables)
for line in fr.readlines():tblName = line.rstrip('\n')# clean old directory in HDFShdfs_command = 'hdfs dfs -rm -r %s/%s/%s' % (dw_parent_dir, tblName, biz_date)# parallel execution import# ${sqoop_import_params} ${sqoop_jdbc_params} --target-dir ${dw_parent_dir}/${p}/${biz_date} --table ${p^^} -m 1 &# sqoopImportCommand = f''' {sqoop_import_params} {sqoop_jdbc_params} --target-dir {dw_parent_dir}/{tblName}/{biz_date} --table {tblName.upper()} -m 1 &'''sqoopImportCommand = '''%s %s --target-dir %s/%s/%s --table %s -m 1 &''' % (sqoop_import_params, sqoop_jdbc_params, dw_parent_dir, tblName, biz_date, tblName.upper())# parallel execution importsubprocess.call(sqoopImportCommand, shell=True)# cur_time=`date "+%F %T"`# cur_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')logging.basicConfig(level=logging.INFO,filename='%s/log/%s_full_imp.log' % (workhome, biz_fmt_date),filemode='a',format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s')# logging.info(cur_time + ' : ' + sqoopImportCommand)logging.info(sqoopImportCommand)time.sleep(15)

导入avro schema

#!/usr/bin/env python
# @Time : 2021/7/20 15:46
# @desc :
__coding__ = "utf-8"
__author__ = "itcast"# import pyhdfs
import logging
import osworkhome = '/opt/sqoop/one_make'
hdfs_schema_dir = '/data/dw/ods/one_make/avsc'
biz_date = '20210101'
biz_fmt_date = '2021-01-01'
local_schema_backup_filename = 'schema_%s.tar.gz' % biz_date
hdfs_schema_backup_filename = '%s/avro_schema_%s.tar.gz' % (hdfs_schema_dir, biz_date)
log_file = '%s/log/upload_avro_schema_%s.log' % (workhome, biz_fmt_date)# append log to file
logging.basicConfig(level=logging.INFO,filename=log_file,filemode='a',format='%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s')os.system('source /etc/profile')
os.system('cd %s' % workhome)#  hadoop fs [generic options] [-test -[defsz] <path>]
# -test -[defsz] <path> :
#   Answer various questions about <path>, with result via exit status.
#     -d  return 0 if <path> is a directory.
#     -e  return 0 if <path> exists.
#     -f  return 0 if <path> is a file.
#     -s  return 0 if file <path> is greater than zero bytes in size.
#     -z  return 0 if file <path> is zero bytes in size, else return 1.
logging.info('Check if the HDFS Avro schema directory %s...', hdfs_schema_dir)
# hdfs = pyhdfs.HdfsClient(hosts="node1,9000", user_name="hdfs")
# print(hdfs.listdir('/'))
# hdfs dfs -test -e ${hdfs_schema_dir} > /dev/null
commStatus = os.system('hdfs dfs -test -e %s > /dev/null' % hdfs_schema_dir)
if commStatus is not 0:logging.info('Path: %s is not exists. Create a new one.', hdfs_schema_dir)logging.info('hdfs dfs -mkdir -p %s', hdfs_schema_dir)os.system('hdfs dfs -mkdir -p %s' % hdfs_schema_dir)logging.info('Check if the file %s/CISS4_CISS_BASE_AREAS.avsc has uploaded to the HFDS...', hdfs_schema_dir)
commStatus = os.system('hdfs dfs -test -e %s/CISS4_CISS_BASE_AREAS.avsc > /dev/null' % hdfs_schema_dir)
if commStatus is not 0:logging.info('Upload all the .avsc schema file.')logging.info('hdfs dfs -put %s/java_code/*.avsc %s', workhome, hdfs_schema_dir)os.system('hdfs dfs -put %s/java_code/*.avsc %s' % (workhome, hdfs_schema_dir))# backup
logging.info('Check if the backup tar.gz file has generated in the local server...')
commStatus = os.system('[ -e %s ]' % local_schema_backup_filename)
if commStatus is not 0:logging.info('package and compress the schema files')logging.info('tar -czf %s ./java_code/*.avsc', local_schema_backup_filename)os.system('tar -czf %s ./java_code/*.avsc' % local_schema_backup_filename)logging.info('Check if the backup tar.gz file has upload to the HDFS...')
commStatus = os.system('hdfs dfs -test -e %s > /dev/null' % hdfs_schema_backup_filename)
if commStatus is not 0:logging.info('upload the schema package file to HDFS')logging.info('hdfs dfs -put %s %s', local_schema_backup_filename, hdfs_schema_backup_filename)os.system('hdfs dfs -put %s %s' %(local_schema_backup_filename, hdfs_schema_backup_filename))

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/337343.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

如何克隆非默认分支

直接git clone下来的我们知道是默认分支&#xff0c;那如何克隆其他分支呢&#xff1a; 比如这个&#xff0c;我们想克隆AdvNet。 我们可以在本地文件夹打开Git Bash 依次输入&#xff1a; git clone --branch AdvNet https://github.com/wgcban/SemiCD.git cd SemiCD git b…

Java基础入门day62

day62 AJAX 概念 AJAX&#xff1a; Asynchronous Javascript And XML AJAX是一种无需重新加载整个网页的情况下&#xff0c;能够更新部分网页的技术 AJAX是一种用于创建快速动态网页的技术 通过在后台与服务器进行少量数据交换&#xff0c;AJAX可以使网页实现异步更新 传统…

源码编译安装LAMP(安装apeche mysql php 论坛 网站 巨详细版)

目录 一.LAMP架构相关概述 1.各组件作用 Linux&#xff08;平台&#xff09; Apache&#xff08;前台&#xff09; MySQL&#xff08;后台&#xff09; PHP/Perl/Python&#xff08;中间连接&#xff09; 总结 二.编译安装Apache httpd服务 1.关闭防火墙&#xff0c;将…

docker一键部署EFK系统(elasticsearch filebeat kibana metricbeat es-head)

EFK日志系统搭建 EFK日志系统介绍功能需求搭建elasticsearch集群规划前提部署核对证书及权限 EFK日志系统介绍 Elasticsearch 是一个实时的、分布式的可扩展的搜索引擎&#xff0c;允许进行全文、结构化搜索&#xff0c;它通常用于索引和搜索大量日志数据&#xff0c;也可用于…

【计算机网络】——物理层(图文并茂)

物理层 一.物理层概述1.物理层要实现的功能2.物理层接口特征1.机械特性2.电气特性3.功能特性4.过程特性 二.物理层下面的传输媒体1.传输媒体的分类2.导向型传输媒体1.同轴电缆2.双绞线3.光纤 3.非导向型传输媒体1.无线电波2.微波3.红外线4.激光5.可见光 三.传输方式1.串行传输与…

Kali : 安装Google Chrome 浏览器和ChromeDriver

目录 一、安装Google Chrome 浏览器 1、下载Google Chrome 2、安装Chrome 3、安装依赖包 二、安装ChromeDriver 1、查看Chrome版本 ​2、下载ChromeDriver 3、解压下载包 4、设置全局访问 5、赋予可执行权限 6、验证chromedriver 7、程序测试 一、安装Google Chrom…

Kafka系列之高频面试题

基础 简介 特点&#xff1a; 高吞吐、低延迟&#xff1a;kafka每秒可以处理几十万条消息&#xff0c;延迟最低只有几毫秒&#xff0c;每个Topic可以分多个Partition&#xff0c;Consumer Group对Partition进行Consumer操作可扩展性&#xff1a;Kafka集群支持热扩展持久性、可…

WEB攻防-JAVAWEB项目常见漏洞

知识点 1.JavaWeb常见安全及代码逻辑 2.目录遍历&身份验证&逻辑&JWT 3.访问控制&安全组件&越权&三方组件 本篇主要了解以上问题在javaweb中的呈现&#xff0c; 第一个重点理解URL与javaweb代码框架的对应方式&#xff0c;java在没有代码的情况下是很难…

【R基础】如何开始学习R-从下载R及Rstudio开始

文章目录 概要下载R流程下载Rstudio流程下载完成-打开 概要 提示&#xff1a;如何开始学习R-从下载R及Rstudio开始&#xff0c;此处我只是想下载指定版本R4.3.3 下载R流程 链接: R官网 文件下载到本地 下载文件展示 按照向导指示安装 下载Rstudio流程 链接: Rstudio官网…

hadoop(1)--hdfs部署(亲测可用)

一、准备&#xff1a; 1、三台集群部署&#xff0c;配置hosts #cat /etc/hosts 192.168.46.128 node1 #nameNode dataNode secondaryNameNode 192.168.46.129 node2 #datanode 192.168.46.130 node3 #datanode说明&#xff1a; NameNode: 主节点管理者 DataNode&…

把自己的服务器添加到presearch节点

Presearch is a scam. Before, judging by the price of the token you should have been able to get between $150-$200 after 12-13 months of regular searches. "If you use this service for the next 11 years you will have earned $30!" Presearch大约需要…

11.1 排序算法

目录 11.1 排序算法 11.1.1 评价维度 11.1.2 理想排序算法 11.1 排序算法 排序算法&#xff08;sorting algorithm&#xff09;用于对一组数据按照特定顺序进行排列。排序算法有着广泛的应用&#xff0c;因为有序数据通常能够被更高效地查找、分析和处理。 如图 1…

STM32—按键控制LED(定时器)

目录 1 、 电路构成及原理图 2 、编写实现代码 main.c exit.c 3、代码讲解 4、烧录到开发板调试、验证代码 5、检验效果 此笔记基于朗峰 STM32F103 系列全集成开发板的记录。 1 、 电路构成及原理图 EXTI&#xff08;External interrupt/event controller&#xff…

Ubuntu22.04之扩展并挂载4T硬盘(二百三十三)

简介&#xff1a; CSDN博客专家&#xff0c;专注Android/Linux系统&#xff0c;分享多mic语音方案、音视频、编解码等技术&#xff0c;与大家一起成长&#xff01; 优质专栏&#xff1a;Audio工程师进阶系列【原创干货持续更新中……】&#x1f680; 优质专栏&#xff1a;多媒…

C 基础 - 预处理命令和基本语法详解

#include <stdio.h> //预处理指令int main() //函数 {printf("Hello, World!"); //输出语句return 0; //返回语句 } 目录 一.预处理指令 1.#define #ifdef #ifndef #if #else #elif #endif 2.#inlcude a.新增一个文件 b.#include c.运行结果 d.扩…

云计算与 openstack

文章目录 一、 虚拟化二、云计算2.1 IT系统架构的发展2.2 云计算2.3 云计算的服务类型 三、Openstack3.1 OpenStack核心组件 一、 虚拟化 虚拟化使得在一台物理的服务器上可以跑多台虚拟机&#xff0c;虚拟机共享物理机的 CPU、内存、IO 硬件资源&#xff0c;但逻辑上虚拟机之…

基于单片机的恒流开关电源 BUCK电路设计

1 前言 1.1课题研究意义 开关电源顾名思义&#xff0c;开关电源便是使用半导体开关器件&#xff08;如晶体管、场效应管、可控硅闸流管等&#xff09;&#xff0c;经过控制电路&#xff0c;使半导体开关器件不停地“导通”和“关闭”&#xff0c;让半导体开关器件对输入的电压…

【数据结构】详解二叉树

文章目录 1.树的结构及概念1.1树的概念1.2树的相关结构概念1.3树的表示1.4树在实际中的应用 2.二叉树的结构及概念2.1二叉树的概念2.2特殊的二叉树2.2.1满二叉树2.2.2完全二叉树 2.3 二叉树的性质2.4二叉树的存储结构2.4.1顺序结构2.4.2链表结构 1.树的结构及概念 1.1树的概念…

if语句知识点

作用 让顺序执行的代码产生分歧。 if 语句 作用&#xff1a;满足条件时&#xff0c;多执行一些代码。 语法&#xff1a; if(bool类型值)//bool类型相关&#xff1a;bool变量&#xff0c;条件运算符表达式&#xff0c;逻辑运算符表达式 {满足条件要执行的代码&#xff0c;写在…

c++ QT 实现QMediaPlayer播放音频显示音频级别指示器

文章目录 效果图概述代码总结 效果图 概述 QMediaPlayer就不介绍了&#xff0c;就提供了一个用于播放音频和视频的媒体播放器 QAudioProbe 它提供了一个探针&#xff0c;用于监控音频流。当音频流被捕获或播放时&#xff0c;QAudioProbe 可以接收到音频数据。这个类在需要访问…