企业大数据基础平台搭建和实用开发代码

目录
理解Hadoop HDFS 6
1 介绍 6
2 HDFS设计原 6
21 设计目标 7
22 系统架构容错性设计 7
23 HDFS适合应类型 7
3 HDFS核心概念 7
31 Blocks 7
32 Namenode & Datanode 7
33 Block Caching 8
34 HDFS Federation 8
35 HDFS HA(High Availability高性) 8
4 命令行接口 9
5 Hadoop文件系统 11
6 Java接口 13
61 读操作 13
62 写数 14
63 目录操作 14
64 删数 15
7 数流（读写流程） 15
71 读文件 15
72 写文件 16
73 致性模型 17
74 Hadoop节点距离 18
8 相关运维工具 18
81 distcp行复制 18
82 衡HDFS集群 19
9 HDFS机架感知概念配置实现 19
机架感知什？ 19
二告诉呢？ 19
三什情况会涉机架感知？ 19
四机架感知需考虑情况（权衡性性带宽消耗） 19
五通什方式够告知Hadoop NameNode Slaves机器属Rack？配置步骤 19
六网络拓扑机器间距离 21
10 HDFS理界面 21
YARN 22
基架构 22
工作机制 22
ResourceManager 23
资源理 23
务调度 23
部结构 23
作业提交全程 24
资源调度器 25
务推测执行 26
YARNWEB UI说明 27
集群运行状态查 29
Hadoop 30 新特性 30
Hadoop Common 30
Hadoop HDFS 30
Hadoop MapReduce 31
Hadoop YARN 31
总结 31
Hive部表外部表区 34
概念理解 34
创建部表t1 34
装载数（t1） 35
创建外部表t2 37
装载数（t2） 37
查文件位置 37
观察HDFS文件 40
重新创建外部表t2 41
官网解释 42
Hive数仓库拉链表流水表全量表增量表 42
Hadoop 320 完全分布式集群搭建 45
集群环境搭建 46
二Hadoop配置修改 47
修改 hadoopenvsh 配置 jdk 路径定义集群操作户 47
修改 coresitexml hadoop核心配置 47
修改 hdfssitexml hadoop 节点配置 47
修改 workers 告知 hadoop hdfsDataNode节点 48
修改 yarnsitexml 配置yarn服务 48
修改mapredsitexml 文件 49
修改配置文件分发节点三台服务器 49
三Hadoop服务启动 49
四运行WordCount 52
Linux台载MySQL 53
yum安装MySQL 56
Hive环境搭建 61
安装 61
二配置理 61
三运行 64
Apache Mahout环境搭建 65
PySpark环境搭建 66
Linux台载Python 38 66
1赖安装 66
2载安装包 67
3解压 67
4安装 67
5添加软链接 67
6测试 67
Linux升级安装python38配置pipyum 67
Linux安装Python 38环境卸载旧Python 70
安装新版Python 2713Python 362（Python 2Python 3存修改默认版Python 362） 72
Linux安装Apache Spark 310详细步骤 73
Spark安装配置 76
Spark集群安装设置 79
Ubuntu 1204Hadoop 220 集群搭建 82
Ubuntu 1404安装Hadoop240（单机模式） 87
启动Spark集群 94
Spark性优化 99
1Spark作业基运行原理 101
2资源参数调优 101
numexecutors 101
executormemory 102
executorcores 102
drivermemory 102
sparkdefaultparallelism 102
sparkstoragememoryFraction 102
sparkshufflememoryFraction 102
3资源参数参考示例 103
4Spark中三种Join策略 103
Broadcast Hash Join 103
Shuffle Hash Join 104
Sort Merge Join 104
5Spark 30 中 AQE新特性 105
6数仓库中数优化般原 109
7Spark 中宽赖窄赖 109
概述 109
详细运行原理 110
8Spark算子 113
9Spark RDD 144
Spark RDD特性 144
Spark RDD核心特性 144
关系型数库数性优化解决方案分表(前表历史表)表分区数清理原 147
原目 147
数否需清理阀值判断 147
满负载周期判断 147
迁移周期判断 147
类型数分区方案 148
历史表清理方案 148
注意点 148
数仓库缓慢变化维(Slow changing demenison) 实现方案 149
MySQLTeradataPySpark代码互转表代码 151
PySpark代码基结构 196
PySparkMySQL导出数parquet文件 197
PySparkTeradata导出数parquet文件 197
PySparkParquent文件写入Hive表 198
PySpark读取HiveSQL查询数写入parquet文件 198
PySpark获取Dataframe采样数保存CSV文件 198
PySpark连接MySQL数库插入数 198
PySpark连接Teradata数库插入数 198
PySpark遍历Dataframe行 199
PySpark移动Parquet文件目录 200
PySpark复制Parquet文件目录 200
PySpark删Parquet文件目录 200
PySpark修改Hive指存储路径 200
PySpark显示HDFS路径文件 201
PySpark显示普通Hive表容量（GB） 201
PySpark显示Hive分区表容量（GB） 201
PySpark显示HDFS目录子目录容量 201
PySpark调SqoopHDFS导入Hive表 201
HiveQLparquet文件创建Hive表 202
HiveQLHive表创建Hive视图 202
HiveQL格式化显示Hive查询结果数 202
Hive导出Hive查询结果数CSV文件 202
HiveQL显示Hive表 202
HiveQL显示Hive数库 202
Shell带日期参数运行HQL脚 202
HiveQL更新视图指天表数 203
HiveQL修改Hive表指存储文件 203
Shell清HDFS里数 204
Shell查HDFS数 204
Sqoop显示MySQL中数库 204
SqoopMySQL数库导入HDFS 204
SqoopHDFS数库导入MySQL 206
Sqoop显示MySQL中数库 206
Teradata支持数类型 207
MySQL支持数类型 209
Hive支持数类型 209
Parquet文件存储格式 210
项目组成 210
数模型 211
StripingAssembly算法 212
Parquet文件格式 215
性 216
项目发展 218
总结 218
Apache Airflow文档 218
原理 221
导 221
Airflow环境安装配置 222
通安装方法 222
Airflow环境安装（Docker） 224
Airflow环境配置（Docker） 224
Airflow环境安装（Windows 10） 225
容 227
项目 227
许证 228
快速开始 231
教程 231
Operator教程 257
图形界面截图 301
概念 305
数分析 323
命令行参数 325
调度触发器 345
插件 348
安全性 351
时区 358
Experimental Rest API 361
整合 362
Lineage 432
常见问题 433
API 参考 436
Apache Airflow 20 新特性 600
TaskFlow API(AIP31) 种新编写dags方式 600
完全REST API(AIP32) 601
调度器性显著提升 602
调度器高兼容 (AIP15) 603
务组 (AIP34) 603
崭新户界面 603
减少传感器负载智传感器 (AIP17) 604
简化KubernetesExecutor 604
Airflow core(核心)providers(第三方安装包) Airflow 拆分 60 包： 604
安全性 605
配置 605
基Apache Airflow企业级数框架架构设计 605

理解Hadoop HDFS
文详细介绍HDFS中许概念理解Hadoop分布式文件系统帮助
1 介绍
现代企业环境中单机容量法存储量数需跨机器存储统理分布集群文件系统称分布式文件系统旦系统中引入网络避免引入网络编程复杂性例挑战果保证节点时候数丢失
传统网络文件系统（NFS）然称分布式文件系统存限制NFS中文件存储单机法提供性保证客户端时访问NFS Server时容易造成服务器压力造成性瓶颈外果NFS中文件进行操作需首先步修改步服务端前客户端见某种程度NFS种典型分布式系统然文件确放远端（单）服务器面

NFS协议栈事实种VFS（操作系统文件种抽象）实现
HDFSHadoop Distributed File System简称Hadoop抽象文件系统种实现Hadoop抽象文件系统系统Amazon S3等集成甚通Web协议（webhsfs）操作HDFS文件分布集群机器时提供副进行容错性保证例客户端写入读取文件直接操作分布集群机器没单点性压力
果零开始搭建完整集群参考[Hadoop集群搭建详细步骤（260）]（httpblogcsdnnetbingduanlbdarticledetails51892750）
2 HDFS设计原
HDFS设计初非常明确应场景适什类型应适什应相明确指导原
21 设计目标
· HDFSHadoop核心项目现数领域事实存储标准高容错性设计运行量廉价商业硬件设计会假设前提
· 首先会首先假设硬件障通常现象说块磁盘障概率非常服务器集群数千台甚万台节点时候（磁盘障）非常司空见惯事情说快速发现定位问题快速做障恢复重设计目标
· 第二HDFS适合容量数流式访问样场景说数场景动辄百G甚T文件相低延时言会更加意批量处理高吞吐样诉求
· 第三形成理念移动计算代价移动数代价数领域普遍识Hadoop推出HDFS存储时候推出MapReduce计算框架出目
· 存储非常文件：里非常指百MG者TB级实际应中已集群存储数达PB级根Hadoop官网YahooHadoop集群约10万颗CPU运行4万机器节点更世界Hadoop集群情况参考Hadoop官网
· 采流式数访问方式 HDFS基样假设：效数处理模式次写入次读取数集常数源生成者拷贝次然做分析工作
分析工作常读取中部分数全部读取整数集需时间读取第条记录延时更重
· 运行商业硬件 Hadoop需特贵reliable（）机器运行普通商机器（家供应商采购）商机器代表低端机器集群中（尤集群）节点失败率较高HDFS目标确保集群节点失败时候会户感觉明显中断
22 系统架构容错性设计
HFDS典型MasterSlave架构里NameNode存储文件系统元数说分块文件路径等等类数BLOCK形式存储DataNode整体HDFS提供client客户提供文件系统命名空间操作说文件开关闭重命名移动等等
说HDFS存储文件提数副机制说里PART 0文件包括Block 1Block 3两Block时设置replica副数等213两Block会复制出外1外3存储外DataNode节点文件拆分两Block存储拥两副里面核心问题拆分数会存储方（DataNode）样调度策略问题Block副存节点（DataBlock）节点挂掉时候数丢失？ HFDS解决问题呢？引入机架感知（Rack Awareness）样概念里引入两机架Rack 1Rack 2台机架会三样节点（DataNode）
23 HDFS适合应类型
场景适合HDFS存储数面列举：
1）低延时数访问
延时求毫秒级应适合采HDFSHDFS高吞吐数传输设计牺牲延时HBase更适合低延时数访问
2）量文件
文件元数（目录结构文件block节点列表blocknode mapping）保存NameNode存中整文件系统文件数量会受限NameNode存
验言文件目录文件块般占150字节元数存空间果100万文件文件占1文件块需约300M存十亿级文件数量现商机器难支持
3）方读写需意文件修改
HDFS采追加（appendonly）方式写入数支持文件意offset修改支持写入器（writer）
3 HDFS核心概念
31 Blocks
物理磁盘中块概念磁盘物理Block磁盘操作单元读写操作均Block单元般512 Byte文件系统物理Block抽象层概念文件系统Block物理磁盘Block整数倍通常KBHadoop提供dffsck类运维工具文件系统Block级进行操作
HDFSBlock块般单机文件系统默认128MHDFS文件拆分成blocksizedchunkchunk作独立单元存储Block文件会占整Block会占实际例果文件1MHDFS中会占1M空间128M
HDFSBlock什？
化查找（seek）时间控制定位文件传输文件时间例假设定位Block需时间10ms磁盘传输速度100Ms果定位Block时间占传输时间例控制1Block需约100M
果Block设置MapReduce务中Map者Reduce务数果集群机器数量会作业运行效率低
Block抽象处
block拆分单文件整磁盘容量构成文件Block分布整集群理单文件占集群中机器磁盘
Block抽象简化存储系统Block需关注权限者等容（容文件级进行控制）
Block作容错高机制中副单元Block单位进行复制
32 Namenode & Datanode
整HDFS集群NamenodeDatanode构成masterworker（）模式Namenode负责构建命名空间理文件元数等Datanode负责实际存储数负责读写工作
Namenode
Namenode存放文件系统树文件目录元数元数持久化2种形式：
· namespcae image
· edit log
持久化数中包括Block节点列表文件Block分布集群中节点信息系统重启时候重新构建（通Datanode汇报Block信息）
HDFS中Namenode成集群单点障Namenode时整文件系统HDFS针单点障提供2种解决机制：
1）备份持久化元数
文件系统元数时写文件系统例时元数写文件系统NFS备份操作步原子
2）Secondary Namenode
Secondary节点定期合Namenodenamespace imageedit log 避免edit log通创建检查点checkpoint合会维护合namespace image副 Namenode完全崩溃时恢复数图Secondary Namenode理界面：

Secondary Namenode通常运行台机器合操作需耗费量CPU存数落NamenodeNamenode完全崩溃时会出现数丢失通常做法拷贝NFS中备份元数Second作新Namenode
HA（High Availability高性）中运行Hot Standby作热备份Active Namenode障代原Namenode成Active Namenode
Datanode
数节点负责存储提取Block读写请求namenode直接客户端数节点周期性Namenode汇报节点存储Block相关信息
33 Block Caching
DataNode通常直接磁盘读取数频繁Block存中缓存默认情况Block数节点会缓存针文件性化配置
作业调度器利缓存提升性例MapReduce务运行Block缓存节点
户者应NameNode发送缓存指令（缓存文件缓存久）缓存池概念理组缓存权限资源
34 HDFS Federation
知道NameNode存会制约文件数量HDFS Federation提供种横扩展NameNode方式Federation模式中NameNode理命名空间部分例NameNode理user目录文件 NameNode理share目录文件
NameNode理namespace volumnvolumn构成文件系统元数NameNode时维护Block Pool保存Block节点映射等信息NameNode间独立节点失败会导致节点理文件
客户端mount table文件路径映射NameNodemount tableNamenode群组封装层层Hadoop文件系统实现通viewfs协议访问
35 HDFS HA(High Availability高性)
HDFS集群中NameNode然单点障（SPOF Single Point Of Failure）元数时写文件系统Second NameNode定期checkpoint利保护数丢失提高性
NameNode唯文件元数fileblock映射负责方挂包括MapReduce作业法进行读写
NameNode障时常规做法元数备份重新启动NameNode元数备份源：
· 文件系统写入中备份
· Second NameNode检查点文件
启动新Namenode需重新配置客户端DataNodeNameNode信息外重启耗时般较久稍具规模集群重启常需十分钟甚数时造成重启耗时原致：
1）元数镜文件载入存耗时较长
2）需重放edit log
3）需收DataNode状态报告满足条件离开安全模式提供写服务
HadoopHA方案
采HAHDFS集群配置两NameNode分处ActiveStandby状态Active NameNode障Standby接责继续提供服务户没明显中断感觉般耗时十秒数分钟
HA涉实现逻辑
1）备需享edit log存储
NameNode命NameNode享份edit log备切换时Standby通回放edit log步数
享存储通常2种选择
· NFS：传统网络文件系统
· QJM：quorum journal manager
QJM专门HDFSHA实现设计提供高edit logQJM运行组journal nodeedit log必须写部分journal nodes通常3节点允许节点失败类似ZooKeeper注意QJM没ZK然HDFS HA确ZK选举Namenode般推荐QJM
2）DataNode需时备发送Block Report
Block映射数存储存中（磁盘）Active NameNode挂掉新NameNode够快速启动需等DatanodeBlock ReportDataNode需时备两NameNode发送Block Report
3）客户端需配置failover模式（失效备援模式户透明）
Namenode切换客户端说感知通客户端库实现客户端配置文件中HDFS URI逻辑路径映射Namenode址客户端会断尝试Namenode址直成功
4）Standby代Secondary NameNode
果没启HAHDFS独立运行守护进程作Secondary Namenode定期checkpoint合镜文件edit日志
果Namenode失败时备份Namenode正关机（停止 Standby）运维员然头启动备份Namenode样没HA时候更省事算种改进重启整程已标准化Hadoop部需运维进行复杂切换操作
NameNode切换通代failover controller实现failover controller种实现默认实现ZooKeeper保证Namenode处active状态
Namenode运行轻量级failover controller进程该进程简单心跳机制监控Namenode存活状态Namenode失败时触发failoverFailover运维手动触发例日常维护中需切换Namenode种情况graceful(优雅) failover非手动触发failover称ungraceful failover
ungraceful failover情况没办法确定失败（判定失败）节点否停止运行说触发failover前Namenode运行QJM次允许Namenode写edit log前Namenode然接受读请求Hadoopfencing杀掉前NamenodeFencing通收回前Namenode享edit log访问权限关闭网络端口原Namenode继续接受服务请求STONITH技术前Namenode关机
HA方案中Namenode切换客户端说见前面已介绍通客户端库完成
4 命令行接口
HDFS提供种交互方式例通Java APIHTTPshell命令行命令行交互通hadoop fs操作例：
1 hadoop fs copyFromLocal 复制文件HDFS
2 hadoop fs mkdir 创建目录
3 hadoop fs ls 列出文件列表
Hadoop中文件目录权限类似POSIX模型包括读写执行3种权限：
· 读权限（r）：读取文件者列出目录中容
· 写权限（w）：文件文件写权限目录写权限指该目录创建者删文件（目录）权限
· 执行权限（x）：文件没谓执行权限忽略目录执行权限访问器目录容
文件目录ownergroupmode三属性owner指文件者group权限组mode 者权限文件属组中组员权限非者非组员权限组成图表示者root拥读写权限supergroup组组员读权限读权限

文件权限否开启通dfspermissionsenabled属性控制属性默认false没开安全限制会客户端做授权校验果开启安全限制会操作文件户做权限校验特殊户superuserNamenode进程标识会针该户做权限校验
ls命令执行结果：

返回结果类似Unix系统ls命令第栏文件moded表示目录紧接着3种权限9位第二栏指文件副数数量通dfsreplication配置目录表示没副说诸者组更新时间文件Unix系统中ls命令致
果需查集群状态者浏览文件目录访问Namenode暴露Http Server查集群信息般namenode机器50070端口

5 Hadoop文件系统
前面Hadoop文件系统概念抽象HDFS中种实现Hadoop提供实现图：

简单介绍Local文件系统抽象hdfs常见两种web形式（webhdfsswebhdfs）实现通HTTP提供文件操作接口harHadoop体系压缩文件文件时候压缩成文件效减少元数数量viewfs前面介绍HDFS Federation张提客户端屏蔽Namenode底层细节ftp顾名思义ftp协议实现文件操作转化ftp协议s3aAmazon云服务提供存储系统实现azure微软云服务台实现
前面提命令行HDFS交互事实方式操作文件系统例Java应程序orgapachehadoopfsFileSystem操作形式操作基FileSystem进行封装里介绍HTTP交互方式
WebHDFSSWebHDFS协议文件系统暴露HTTP操作种交互方式原生Java客户端慢适合操作文件通HTTP2种访问方式直接访问通代理访问
直接访问
直接访问示意图：

NamenodeDatanode默认开嵌入式web serverdfswebhdfsenabled默认truewebhdfs通服务器交互元数操作通namenode完成文件读写首先发namenode然重定datanode读取（写入）实际数流
通HDFS代理

采代理示意图示代理处通代理实现负载均衡者带宽进行限制者防火墙设置代理通HTTP者HTTPS暴露WebHDFS应webhdfsswebhdfs URL Schema
代理作独立守护进程独立namenodedatanodehttpfssh脚默认运行14000端口
FileSystem直接操作命令行HTTTP外C语言APINFSFUSER等方式里做介绍
6 Java接口
实际应中HDFS数操作通FileSystem操作部分重点介绍相关接口关注HDFS实现类DistributedFileSystem相关类
61 读操作
URL读取数者直接FileSystem操作
Hadoop URL读取数
javanetURL类提供资源定位统抽象定义种URL Schema提供相应处理类进行实际操作hdfs schema便样种实现
1 InputStream in null
2 try {
3 in new URL(hdfsmasteruserhadoop)openStream()
4 }finally{
5 IOUtilscloseStream(in)
6 }
定义Schema需设置URLStreamHandlerFactory操作JVM进行次次操作会导致通常静态块中完成面截图示例：

FileSystem API读取数
1）首先获取FileSystem实例般静态get工厂方法
1 public static FileSystem get(Configuration conf) throws IOException
2 public static FileSystem get(URI uri Configuration conf) throws IOException
3 public static FileSystem get(URI uri Configuration confString user) throws IOException
果文件通getLocal获取文件系统象：
public static LocalFileSystem getLocal(COnfiguration conf) thrown IOException
2）调FileSystemopen方法获取输入流
1 public FSDataInputStream open(Path f) throws IOException
2 public abstarct FSDataInputStream open(Path f int bufferSize) throws IOException
默认情况open4KBBuffer根需行设置
3）FSDataInputStream进行数操作
FSDataInputStreamjavaioDataInputStream特殊实现基础增加机读取部分读取力
1 public class FSDataInputStream extends DataInputStream
2 implements Seekable PositionedReadable
3 ByteBufferReadable HasFileDescriptor CanSetDropBehind CanSetReadahead
4 HasEnhancedByteBufferAccess
机读取操作通Seekable接口定义：
1 public interface Seekable {
2 void seek(long pos) throws IOException
3 long getPos() throws IOException
4 }
seek操作开销昂贵慎
部分读取通PositionedReadable接口定义：
1 public interface PositionedReadable{
2 public int read(long pistion byte[] bufferint offser int length) throws IOException
3 public int readFully(long pistion byte[] bufferint offser int length) throws IOException
4 public int readFully(long pistion byte[] buffer) throws IOException
5 }
62 写数
HDFS中文件FileSystem类create方法重载形式创建create方法返回输出流FSDataOutputStream调返回输出流getPos方法查前文件位移进行seek操作HDFS仅支持追加操作
创建时传递回调接口Peofressable获取进度信息
append(Path f)方法追加容已文件实现提供该方法例Amazon文件实现没提供追加功
面例子：
1 String localSrc args[0]
2 String dst args[1]
3
4 InputStream in new BufferedInputStream(new FileInputStream(localSrc))
5
6 Configuration conf new Configuration()
7 FileSystem fs FileSystemget(URIcreate(dst)conf)
8
9 OutputStream out fscreate(new Path(dst) new Progressable(){
10 public vid progress(){
11 Systemoutprint()
12 }
13 })
14
15 IOUtilscopyBytes(in out 4096true)
63 目录操作
mkdirs（）方法会动创建没级目录
HDFS中元数封装FileStatus类中包括长度block sizereplicaions修改时间者权限等信息FileSystem提供getFileStatus方法获取FileStatusexists()方法判断文件者目录否存
列出文件（list）listStatus方法查文件者目录信息
1 public abstract FileStatus[] listStatus(Path f) throws FileNotFoundException
2 IOException
Path文件时候返回长度1数组FileUtil提供stat2Paths方法FileStatus转化Path象
globStatus通配符文件路径进行匹配：
public FileStatus[] globStatus(Path pathPattern) throws IOException
· 1
PathFilter定义文件名滤根文件属性进行滤类似javaioFileFilter例面例子排定正表达式文件：
1 public interfacePathFilter{
2 boolean accept(Path path)
3 }
64 删数
FileSystemdelete()方法
public boolean delete(Path f boolean recursive) throws IOException
· 1
recursive参数f文件时候忽略果f文件recursicetrue删整目录否抛出异常
7 数流（读写流程）
接详细介绍HDFS读写数流程致性模型相关概念
71 读文件
致读文件流程：

1）客户端传递文件PathFileSystemopen方法
2）DFS采RPC远程获取文件开始blockdatanode址Namenode会根网络拓扑结构决定返回节点（前提节点block副）果客户端身Datanode节点刚block副直接读取
3）客户端open方法返回FSDataInputStream象读取数（调read方法）
4）DFSInputStream（FSDataInputStream实现改类）连接持第block节点反复调read方法读取数
5）第block读取完毕寻找block佳datanode读取数果必DFSInputStream会联系Namenode获取批Block 节点信息(存放存持久化）寻址程客户端见
6）数读取完毕客户端调close方法关闭流象
读数程中果Datanode通信发生错误DFSInputStream象会尝试佳节点读取数记住该失败节点续Block读取会连接该节点
读取BlockDFSInputStram会进行检验验证果Block损坏尝试节点读取数损坏block汇报Namenode
客户端连接datanode获取数namenode指导样支持量发客户端请求namenode流量均匀分布整集群
Block位置信息存储namenode存中相应位置请求非常高效会成瓶颈
72 写文件

步骤分解
1）客户端调DistributedFileSystemcreate方法
2）DistributedFileSystem远程RPC调Namenode文件系统命名空间中创建新文件时该文件没关联block 程中Namenode会做校验工作例否已存名文件否权限果验证通返回FSDataOutputStream象果验证通抛出异常客户端
3）客户端写入数时候DFSOutputStream分解packets（数包）写入数队列中该队列DataStreamer消费
4）DateStreamer负责请求Namenode分配新block存放数节点节点存放Block副构成道 DataStreamerpacket写入道第节点第节点存放packet转发节点节点存放继续传递
5）DFSOutputStream时维护ack queue队列等datanode确认消息道datanode确认packetack队列中移
6）数写入完毕客户端close输出流packet刷新道中然安心等datanode确认消息全部确认告知Namenode文件完整 Namenode时已知道文件Block信息（DataStreamer请求Namenode分配block）需等达副数求然返回成功信息客户端
Namenode决定副存Datanode？
HDFS副存放策略性写带宽读带宽间权衡默认策略：
· 第副放客户端相机器果机器集群外机选择（会选择容量太慢者前操作太繁忙）
· 第二副机放第副机架
· 第三副放第二副机架节点满足条件节点中机选择
· 更副整集群机选择然会量避免太副机架
副位置确定建立写入道时候会考虑网络拓扑结构面存放策略：

样选择滴衡性读写性
· 性：Block分布两机架
· 写带宽：写入道程需跨越交换机
· 读带宽：两机架中选读取
73 致性模型
致性模型描述文件系统中读写操见性HDFS中文件旦创建文件系统命名空间中见：
1 Path p new Path(p)
2 fscreate(p)
3 assertTaht(fsexists(p)is(true))
写入文件容保证见象流已刷新
`java
Path p new Path(p)
OutputStream out fscreate(p)
outwrite(contentgetBytes(UTF8))
outflush()
assertTaht(fsgetFileStatus(p)getLen0L) 0调flush
1
2 果需强制刷新数DatanodeFSDataOutputStreamhflush方法强制缓刷datanode
3 hflushHDFS保证时间点止写入文件数达数节点
4 ```java
5 Path p new Path(p)
6 OutputStream out fscreate(p)
7 outwrite(contentgetBytes(UTF8))
8 outflush()
9 assertTaht(fsgetFileStatus(p)getLenis(((longcontentlength())))
关闭象流时部会调hflush方法hflush保证datanode数已写入磁盘保证写入datanode存机器断电时候导致数丢失果保证写入磁盘hsync方法hsync类型fsync（）系统调fsync提交某文件句柄缓数
1 FileOutputStreamout new FileOutPutStream(localFile)
2 outwrite(contentgetBytes(UTF8))
3 outflush()
4 outgetFD()sync()
5 assertTaht(localFilegetLenis(((longcontentlength())))
hflushhsync会导致吞吐量降设计应时需吞吐量数健壮性间做权衡
外文件写入程中前正写入BlockReader见
74 Hadoop节点距离
读取写入程中namenode分配Datanode时候会考虑节点间距离HDFS中距离没
采带宽衡量实际中难准确度量两台机器间带宽
Hadoop机器间拓扑结构组织成树结构达公父节点需跳转数作距离事实距离矩阵例子面例子简明说明距离计算：

数中心机架节点距离0
数中心机架节点距离2
数中心机架节点距离4
数中心机架节点距离6

Hadoop集群拓扑结构需手动配置果没配置Hadoop默认节点位数中心机架
8 相关运维工具
81 distcp行复制
前面关注点单线程访问果需行处理文件需编写应Hadoop提供distcp工具行导入数Hadoop者Hadoop导出例子：
1 hadoop distcp file1 file2 作fs cp命令高效代
2 hadoop distcp dir1 dir2
3 hadoop distcp update dir1 dir2 #update参数表示步更新文件保持变
distcp底层MapReduce实现map实现没reducemap中行复制文件 distcpmap间均分配文件map数量通m参数指定
hadoop distcp update delete p hdfsmaster19000foo hdfsmaster2foo
样操作常两集群间复制数update参数表示步更新数delete会删目标目录中存源目录存文件p参数表示保留文件全校block副数量等属性
果两集群Hadoop版兼容webhdfs协议：
hadoop distcp webhdfsnamenode150070foo webhdfsnamenode250070foo
82 衡HDFS集群
distcp工具中果指定map数量1仅速度慢Block第副全部落运行唯map节点直磁盘溢出distcp时候默认map数量20
HDFSBlock均匀分布节点时候工作果没办法作业中量保持集群衡例限制map数量（便节点作业）balancer工具调整集群Block分布
9 HDFS机架感知概念配置实现
机架感知什？
告诉 Hadoop 集群中台机器属机架
二告诉呢？
Hadoop机架感知非适应Hadoop集群分辨某台Slave 机器属Rack非智感知需 Hadoop理者告知 Hadoop台机器属Rack样HadoopNameNode启动初始化时会机器 rack 应信息保存存中作接 HDFS 写块操作分配 datanode
列表时（ 3 block 应三台 datanode）选择 datanode 策略量三副分布 rack
三什情况会涉机架感知？
Hadoop 集群规模情况
四机架感知需考虑情况（权衡性性带宽消耗）
（1）节点间通信够量发生机架跨机架
（2）提高容错力NameNode会数块副放机架
五通什方式够告知Hadoop NameNode Slaves机器属Rack？配置步骤
1默认情况Hadoop机架感知没启通常情况Hadoop集群 HDFS 选机器时候机选择说写数时Hadoop第块数 block1写rack1然机选择block2 写入rack2 时两Rack间产生数传输流量接机情况block3 重新写回 rack1时两Rack间产生次数流量Job处理数量非常者Hadoop推送数量非常时候种情况会造成 Rack间网络流量成倍升成性瓶颈进影响作业性整集群服务
Hadoop机架感知功启配置非常简单NameNode机器 hadoopsitexml 配置文件中配置选项：
topologyscriptfilenamepathtoRackAwarepy
配置选项 value 指定执行程序通常脚该脚接受参数输出值接受参数通常某台 DataNode机器IP址输出值通常该IP址应DataNodeRack例rack1NameNode启动时会判断该配置选项否空果非空表示已机架感知配置时NameNode会根配置寻找该脚接收 DataNodeHeartbeat时该 DataNodeIP址作参数传该脚运行输出作该DataNode属机架保存存Map中
脚编写需真实网络拓朴机架信息解清楚通该脚够机器IP址正确映射相应机架
简单实现：
#usrbinpython
#codingUTF8 –
import sys
rack {hadoopnode176tjrack1
hadoopnode178tjrack1
hadoopnode179tjrack1
hadoopnode180tjrack1
hadoopnode186tjrack2
hadoopnode187tjrack2
hadoopnode188tjrack2
hadoopnode190tjrack2
192168115rack1
192168117rack1
192168118rack1
192168119rack1
192168125rack2
192168126rack2
192168127rack2
192168129rack2
}
if namemain
print( + rackget(sysargv[1] rack0)
没确切文档说明底机名 ip 址会传入脚脚中兼容机名 ip 址果机房架构较复杂话脚返回：dc1rack1 类似字符串
执行命令：chmod +x RackAwarepy
重启NameNode果配置成功NameNode启动日志中会输出：
INFO orgapachehadoopnetNetworkTopology Adding a new node rack119216811550010
六网络拓扑机器间距离
里基网络拓扑案例介绍复杂网络拓扑中Hadoop集群台机器间距离

机架感知NameNode 画出图示DataNode网络拓扑图D1R1 交换机底层 datanode H1
rackidD1R1H1H1 parent R1R1 D1 rackid信息通 topologyscriptfilename 配置 rackid 信息计算出意两台DataNode间距离
1 distance(D1R1H1D1R1H1)0 相DataNode
2 distance(D1R1H1D1R1H2)2 RackDataNode
3 distance(D1R1H1D1R1H4)4 IDC（互联网数中心（机房））DataNode
4 distance(D1R1H1D2R3H7)6 IDCDataNode
10 HDFS理界面
里HDFS理情况hadoop001开浏览器进入HDFS理界面输入：19216821612850070

点击DataNodes

YARN
Yarn资源调度台负责运算程序提供服务器运算资源相分布式操作系统台MapReduce等运算程序相运行操作系统应程序
基架构
  YARNResourceManagerNodeManagerApplicationMasterContainer等组件构成

工作机制
1）运行机制

2）工作机制详解
（0）MapReduce程序提交客户端节点
（1）Yarn RunnerResourceManager申请Application
（2）ResourceManager该应程序资源路径返回Yarn Runner
（3）该程序运行需资源提交HDFS
（4）程序资源提交完毕申请运行MapReduce Application Master
（5）RM户请求初始化成Task
（6）中NodeManager领取Task务
（7）该NodeManager创建容器Container产生MapReduce Application Master
（8）ContainerHDFS拷贝资源
（9）MapReduce Application MasterResourceManager申请运行Map Task资源
（10）ResourceManager运行Map Task务分配外两NodeManager两NodeManager分领取务创建容器
（11）MapReduce两接收务NodeManager发送程序启动脚两NodeManager分启动Map TaskMap Task数分区排序
（12）MapReduce Application Master等Map Task运行完毕RM申请容器运行Reduce Task
（13）Reduce TaskMap Task获取相应分区数
（14）程序运行完毕MapReduce会ResourceManager申请注销
ResourceManager
负责全局资源理务调度整集群成计算资源池关注分配应负责容错
资源理
1 前资源节点分成Map slotReduce slot现ContainerContainer根需运行ApplicationMasterMapReduce者意程序
2 前资源分配静态目前动态资源利率更高
3 Container资源申请单位资源申请格式： resourcename：机名机架名*（代表意机器） resourcerequirement：目前支持CPU存
4 户提交作业ResourceManager然某NodeManager分配Container运行ApplicationMasterApplicationMaster根身程序需ResourceManager申请资源
5 YARN套Container生命周期理机制ApplicationMasterContainer间理应程序定义
务调度
1 关注资源情况根需求合理分配资源
2 Scheluer根申请需特定机器申请特定资源（ApplicationMaster负责申请资源时数化考虑ResourceManager量满足申请需求指定机器分配Container减少数移动）
部结构

· Client Service 应提交终止输出信息（应队列集群等状态信息）
· Adaminstration Service 队列节点Client权限理
· ApplicationMasterService 注册终止ApplicationMaster 获取ApplicationMaster资源申请取消请求异步传Scheduler 单线程处理
· ApplicationMaster Liveliness Monitor 接收ApplicationMaster心跳消息果某ApplicationMaster定时间没发送心跳务失效资源会回收然ResourceManager会重新分配ApplicationMaster运行该应（默认尝试2次）
· Resource Tracker Service 注册节点接收注册节点心跳消息
· NodeManagers Liveliness Monitor 监控节点心跳消息果长时间没收心跳消息认该节点效时该节点Container标记成效会调度务该节点运行
· ApplicationManager 理应程序记录理已完成应
· ApplicationMaster Launcher 应提交负责NodeManager交互分配Container加载ApplicationMaster负责终止销毁
· YarnScheduler 资源调度分配 FIFO(with Priority)FairCapacity方式
· ContainerAllocationExpirer 理已分配没启Container超定时间回收
作业提交全程
1）作业提交程YARN

作业提交全程详解
（1）作业提交
第0步：client调jobwaitForCompletion方法整集群提交MapReduce作业
第1步：clientRM申请作业id
第2步：RMclient返回该job资源提交路径作业id
第3步：client提交jar包切片信息配置文件指定资源提交路径
第4步：client提交完资源RM申请运行MrAppMaster
（2）作业初始化
第5步：RM收client请求该job添加容量调度器中
第6步：某空闲NM领取该job
第7步：该NM创建Container产生MRAppmaster
第8步：载client提交资源
（3）务分配
第9步：MrAppMasterRM申请运行maptask务资源
第10步：RM运行maptask务分配外两NodeManager两NodeManager分领取务创建容器
（4）务运行
第11步：MR两接收务NodeManager发送程序启动脚两NodeManager分启动maptaskmaptask数分区排序
第12步：MrAppMaster等maptask运行完毕RM申请容器运行reduce task
第13步：reduce taskmaptask获取相应分区数
第14步：程序运行完毕MR会RM申请注销
（5）进度状态更新
YARN中务进度状态(包括counter)返回应理器客户端秒(通mapreduceclientprogressmonitorpollinterval设置)应理器请求进度更新展示户
（6）作业完成
应理器请求作业进度外客户端5分钟会通调waitForCompletion()检查作业否完成时间间隔通mapreduceclientcompletionpollinterval设置作业完成应理器container会清理工作状态作业信息会作业历史服务器存储备户核查
2）作业提交程MapReduce

资源调度器
目前Hadoop作业调度器三种：FIFOCapacity SchedulerFair SchedulerHadoop272默认资源调度器Capacity Scheduler
具体设置详见：yarndefaultxml文件

    The class to use as the resource scheduler
    yarnresourcemanagerschedulerclass
orgapachehadoopyarnserverresourcemanagerschedulercapacityCapacityScheduler

1）先进先出调度器（FIFO）

2）容量调度器（Capacity Scheduler）

3）公调度器（Fair Scheduler）

务推测执行
1）作业完成时间取决慢务完成时间
作业干Map务Reduce务构成硬件老化软件Bug等某务运行非常慢
典型案例：系统中99Map务完成少数Map老进度慢完成办？
2）推测执行机制：
发现拖腿务某务运行速度远慢务均速度拖腿务启动备份务时运行谁先运行完采谁结果
3）执行推测务前提条件
（1）task备份务
（2）前job已完成task必须005（5）
（3）开启推测执行参数设置Hadoop272 mapredsitexml文件中默认开

  mapreducemapspeculative
  true
  If true then multiple instances of some map tasks                may be executed in parallel

  mapreducereducespeculative
  true
  If true then multiple instances of some reduce tasks
               may be executed in parallel

4）启推测执行机制情况
   （1）务间存严重负载倾斜
   （2）特殊务务数库中写数
5）算法原理

YARNWEB UI说明
安装完Yarn浏览器中通httpmaster8088访问YarnWEB UI图：

详细解释图中标记1(cluster)2(Nodes)两界面中资源关信息

面7字段信息进行解释：
1Active Nodes：表示Yarn集群理节点数实NodeManager数集群2NodeManager
2Memory Total：表示Yarn集群理存总存总等NodeManager理存NodeManager理存通yarnsitexml中配置进行配置：

yarnnodemanagerresourcememorymb
1630
表示NodeManager理存

配置中NodeManager理存1630MB整Yarn集群理存总1630MB * 2 3260MB约等318GBMemory Total
3Vcores Total：表示Yarn集群理cpu虚拟核心总数等NodeManager理虚拟核心NodeManager理虚拟核心数通yarnsitexml中配置进行配置

yarnnodemanagerresourcecpuvcores
2
表示NodeManager理虚拟核心数

配置中NodeManager理虚拟核心数2整Yarn集群理虚拟核心总数2 * 2 4Vcores Total
4Scheduler Type：表示资源分配类型Hadoopyarn安装文章中说三中资源调度
5Minimum Allocation：分配资源说务Yarn申请资源时候Yarn少会分配资源务分配存核心数分配置yarnschedulerminimumallocationmb(默认值1024MB)yarnschedulerminimumallocationvcores(默认值1)控制
6Maximum Allocation：分配资源说务Yarn申请资源时候Yarn会分配资源务分配存核心数分配置yarnschedulermaximumallocationmb(默认值8192MB)yarnschedulermaximumallocationvcores(默认值32)控制然两值肯定集群理资源

面Yarn集群理两NodeManager状态信息分：
1Rack：表示NodeManager机器机架
2Node State：表示NodeManager状态
3Mem Used：表示NodeManager已存Mem Avail：表示NodeManager剩少存VCores Used：表示NodeManager已VCores数量VCores Avail：表示NodeManager剩少VCores数量
点击Node Address

进入界面：

界面信息slave2NodeManager详细信息中Total Vmem allocated for Containers表示NodeManager理虚拟存虚拟存yarnsitexml中配置设置：

yarnnodemanagervmempmemratio
41
表示NodeManager理虚拟存物理存例

面配置yarnnodemanagervmempmemratio虚拟存物理存例41说虚拟存物理存41倍虚拟存1630MB * 41 6683MB约等653GB
集群运行状态查

注：般资源超配置资源话  Staday Fair Shar mem  Min Resources mem

发生然Staday Fair Shar mem  Min Resources mem

暂时未遇单列超max资源配情况Staday Fair Shar mem Min Resources mem情况

Hadoop 30 新特性
Hadoop 30功性方面hadoop核进行项重改进包括：
Hadoop Common
（1）精简Hadoop核包括剔期API实现默认组件实现换成高效实现（FileOutputCommitter缺省实现换v2版废hftp转webhdfs代移Hadoop子实现序列化库orgapachehadoopRecords
（2）Classpath isolation防止版jar包突google Guava混合HadoopHBaseSpark时容易产生突（httpsissuesapacheorgjirabrowseHADOOP11656）
（3）Shell脚重构 Hadoop 30Hadoop理脚进行重构修复量bug增加新特性支持动态命令等httpsissuesapacheorgjirabrowseHADOOP9902
Hadoop HDFS
（1）HDFS支持数擦编码HDFS降低性前提节省半存储空间（httpsissuesapacheorgjirabrowseHDFS7285）
（2）NameNode支持支持集群中activestandby namenode部署方式注：ResourceManager特性hadoop 20中已支持（httpsissuesapacheorgjirabrowseHDFS6440）
Hadoop MapReduce
（1）Tasknative优化MapReduce增加CC++map output collector实现（包括SpillSortIFile等）通作业级参数调整切换该实现shuffle密集型应性提高约30（httpsissuesapacheorgjirabrowseMAPREDUCE2841）
（2）MapReduce存参数动推断Hadoop 20中MapReduce作业设置存参数非常繁琐涉两参数：mapreduce{mapreduce}memorymbmapreduce{mapreduce}javaopts旦设置合理会存资源浪费严重前者设置4096MB者Xmx2g剩余2g实际法java heap（httpsissuesapacheorgjirabrowseMAPREDUCE5785）
Hadoop YARN
（1）基cgroup存隔离IO Disk隔离（httpsissuesapacheorgjirabrowseYARN2619）
（2）curator实现RM leader选举（httpsissuesapacheorgjirabrowseYARN4438）
（3）containerresizing（httpsissuesapacheorgjirabrowseYARN1197）
（4）Timelineserver next generation （httpsissuesapacheorgjirabrowseYARN2928）
hadoop30新参数
hadoop30
HADOOP
Move to JDK8+
Classpath isolation on by default HADOOP11656
Shell script rewrite HADOOP9902
Move default ports out of ephemeral range HDFS9427
HDFS
Removal of hftp in favor of webhdfs HDFS5570
Support for more than two standby NameNodes HDFS6440
Support for Erasure Codes in HDFS HDFS7285
YARN
MAPREDUCE
Derive heap size or mapreduce*memorymb automatically MAPREDUCE5785
HDFS7285中实现Erasure Coding新功鉴功远没发布阶段面块相关代码会进行进步改造做谓预分析帮助家提前解Hadoop社区目前实现功前没接触Erasure Coding技术中间程确实偶然相信文带家收获
Erasure coding纠删码技术简称EC种数保护技术早通信行业中数传输中数恢复种编码容错技术通原始数中加入新校验数部分数产生关联性定范围数出错情况通纠删码技术进行恢复面结合图片进行简单演示首先原始数n然加入m校验数块图示
Parity部分校验数块行数块组成Stripe条带行条带n数块m校验块组成原始数块校验数块通现数块进行恢复原
果校验数块发生错误通原始数块进行编码重新生成果原始数块发生错误通校验数块解码重新生成
mn值固定变进行相应调整会奇中底什原理呢实道理简单面图成矩阵矩阵运算具逆性数进行恢复出张标准矩阵相图家二者关联
总结
Hadoop 30alpha版预计2016夏天发布GA版11月12月发布
Hadoop 30中引入重功优化包括HDFS 擦编码Namenode支持MR Native Task优化YARN基cgroup存磁盘IO隔离YARN container resizing等

相前生产发布版Hadoop 2Apache Hadoop 3整合许重增强功 Hadoop 3版提供稳定性高质量API实际产品开发面简介绍Hadoop3变化
· 低Java版求Java7变Java8
    Hadoopjar基Java 8运行版进行编译执行Java 7更低Java版户需升级Java 8
· HDFS支持纠删码（erasure coding）
    纠删码种副存储更节省存储空间数持久化存储方法ReedSolomon(104)标准编码技术需14倍空间开销标准HDFS副技术需3倍空间开销纠删码额外开销重建远程读写通常存储常数（冷数）外新特性时户需考虑网络CPU开销
· YARN时间线服务 v2(YARN Timeline Service v2)
    YARN Timeline Service v2应两挑战：（1）提高时间线服务扩展性性（2）通引入流(flow)聚合(aggregation)增强性代Timeline Service v1xYARN Timeline Service v2 alpha 2提出样户开发者进行测试提供反馈建议YARN Timeline Service v2测试容器中
· 重写Shell脚
    Hadoopshell脚重写修补许长期存bug增加新特性
· 覆盖客户端jar（Shaded client jars）
    2x版中hadoopclient Maven artifact配置会拉取hadoop传递赖hadoop应程序环境变量回带传递赖版应程序版相突问题
    HADOOP11804 添加新 hadoopclientapihadoopclientruntime artifcathadoop赖隔离单Jar包中避免hadoop赖渗透应程序类路径中
· 支持Opportunistic ContainersDistributed Scheduling
    ExecutionType概念引入样应够通Opportunistic执行类型请求容器调度时没资源种类型容器会分发NM中执行程序种情况容器放入NM队列中等资源便执行Opportunistic container优先级默认Guaranteedcontainer低需情况资源会抢占便Guaranteed container样需提高集群率
    Opportunistic container默认中央RM分配目前已增加分布式调度器支持该分布式调度器做AMRProtocol解析器实现
· MapReduce务级优化
    MapReduce添加映射输出收集器化实现支持密集型洗牌操作（shuffleintensive）jobs带30性提升
· 支持余2NameNodes
    针HDFS NameNode高性初实现方式提供活跃（active）NameNode备（Standby）NameNode通3JournalNode法定数量复制编辑种架构够系统中节点障进行容错
    该功够通运行更备NameNode提供更高容错性满足部署需求通配置3NameNode5JournalNode集群够实现两节点障容错
· 修改重服务默认端口
    前Hadoop版中重Hadoop服务默认端口Linux时端口范围容（3276861000）意味着启动程中服务器端口突会启动失败突端口已时端口范围移NameNodeSecondary NameNodeDataNodeKMS会受影响文档已做相应修改通阅读发布说明 HDFS9427HADOOP12811详细解修改端口
· 提供文件系统连接器（filesystem connnector）支持Microsoft Azure Data LakeAliyun象存储系统
    Hadoop支持Microsoft Azure Data LakeAliyun象存储系统集成作Hadoop兼容文件系统
· 数节点置衡器（Intradatanode balancer）
    单DataNode理磁盘情况执行普通写操作时磁盘量较均添加者更换磁盘时会导致DataNode磁盘量严重均衡目前HDFS均衡器关注点DataNode间（inter）intra处理种均衡情况
    hadoop3 中通DataNode部均衡功已处理述情况通hdfs diskbalancer ClI调
· 重写守护进程务堆理机制
    针Hadoop守护进程MapReduce务堆理机制Hadoop3 做系列修改
    HADOOP10950 引入配置守护进程堆新方法特HADOOP_HEAPSIZE配置方式已弃根机存进行动调整
    MAPREDUCE5785 简化MAP配置减少务堆需务配置Java选项中明确指出需堆已明确指出堆现配置会受该改变影响
· S3GuradS3A文件系统客户端提供致性元数缓存
    HADOOP13345 亚马逊S3存储S3A客户端提供选特性：够DynamoDB表作文件目录元数快速致性存储
· HDFS基路器互联（HDFS RouterBased Federation）
    HDFS RouterBased Federation添加RPC路层HDFS命名空间提供联合视图现ViewFsHDFS Federation功类似区通服务端理表加载原客户端理简化现存HDFS客户端接入federated cluster操作
· 基API配置Capacity Scheduler queue configuration
    OrgQueue扩展capacity scheduler提供种编程方法该方法提供REST API修改配置户通远程调修改队列配置样队列administer_queue ACL理员实现动化队列配置理
· YARN资源类型
    Yarn资源模型已般化支持户定义计算资源类型仅仅CPU存集群理员定义GPU数量软件序列号连接存储资源然Yarn务够资源进行调度

Hive部表外部表区
未external修饰部表（managed table）external修饰外部表（external table）
区：
部表数Hive身理外部表数HDFS理
部表数存储位置hivemetastorewarehousedir（默认：userhivewarehouse）外部表数存储位置制定（果没LOCATIONHiveHDFSuserhivewarehouse文件夹外部表表名创建文件夹属表数存放里）
删部表会直接删元数（metadata）存储数删外部表仅仅会删元数HDFS文件会删
部表修改会修改直接步元数外部表表结构分区进行修改需修复（MSCK REPAIR TABLE table_name）
进行试验进行理解
概念理解
创建部表t1
create table t1(
id int
name string
hobby array
add map
)
row format delimited
fields terminated by ＇＇
collection items terminated by ＇＇
map keys terminated by ＇＇

2 查表描述：desc t1

装载数（t1）
注：般少insert （insert overwrite）语句算算插入条数会调MapReduce里选择Load Data方式
LOAD DATA [LOCAL] INPATH ＇filepath＇ [OVERWRITE] INTO TABLE tablename [PARTITION (partcol1val1 partcol2val2 )]
创建文件粘贴述记录载图：

文件容
1xiaomingbookTVcodebeijingchaoyangshagnhaipudong
2lileibookcodenanjingjiangningtaiwantaibei
3lihuamusicbookheilongjianghaerbin
然载
load data local inpath ＇homehadoopDesktopdata＇ overwrite into table t1
忘记写文件名data笔者第次忘记写整Desktop传查全null乱码
查表容：
select * from t1

创建外部表t2
create external table t2(
id int
name string
hobby array
add map
)
row format delimited
fields terminated by ＇＇
collection items terminated by ＇＇
map keys terminated by ＇＇
location ＇usert2＇

装载数（t2）
load data local inpath ＇homehadoopDesktopdata＇ overwrite into table t2

查文件位置
图NameNode50070explorerhtml#user目录t2文件

t1呢？前配置默认路径里

样通命令行获两者位置信息：
desc formatted table_name

注：图中managed table部表external table外部表
##分删部表外部表
面分删部表外部表查区

观察HDFS文件
发现t1已存

t2然存

外部表仅仅删元数
重新创建外部表t2
create external table t2(
id int
name string
hobby array
add map
)
row format delimited
fields terminated by ＇＇
collection items terminated by ＇＇
map keys terminated by ＇＇
location ＇usert2＇

里面插入数select * 结果

见数然
官网解释
官网中关external表介绍：
A table created without the EXTERNAL clause is called a managed table because Hive manages its data
Managed and External Tables
By default Hive creates managed tables where files metadata and statistics are managed by internal Hive processes A managed table is stored under the hivemetastorewarehousedir path property by default in a folder path similar to appshivewarehousedatabasenamedbtablename The default location can be overridden by the location property during table creation If a managed table or partition is dropped the data and metadata associated with that table or partition are deleted If the PURGE option is not specified the data is moved to a trash folder for a defined duration
Use managed tables when Hive should manage the lifecycle of the table or when generating temporary tables
An external table describes the metadata schema on external files External table files can be accessed and managed by processes outside of Hive External tables can access data stored in sources such as Azure Storage Volumes (ASV) or remote HDFS locations If the structure or partitioning of an external table is changed an MSCK REPAIR TABLE table_name statement can be used to refresh metadata information
Use external tables when files are already present or in remote locations and the files should remain even if the table is dropped
Managed or external tables can be identified using the DESCRIBE FORMATTED table_name command which will display either MANAGED_TABLE or EXTERNAL_TABLE depending on table type
Statistics can be managed on internal and external tables and partitions for query optimization
Hive官网介绍：
httpscwikiapacheorgconfluencedisplayHiveLanguageManual+DDL#LanguageManualDDLDescribeTableViewColumn
Hive数仓库拉链表流水表全量表增量表
1 全量表：天新状态数
2 增量表：天新增数增量数次导出新数
3 拉链表：维护历史状态新状态数种表拉链表根拉链粒度实际相快做优化部分变记录已通拉链表方便原出拉链时点客户记录
4 流水表：表修改会记录反映实际记录变更

拉链表通常账户信息历史变动进行处理保留结果流水表天交易形成历史
流水表统计业务相关情况拉链表统计账户客户情况
数仓库拉链表（原理设计Hive中实现）

情况保持历史状态需拉链表做样做目保留状态情况节省空间

拉链表适种情况吧

数量点表中某字段变化呢变化频率高业务需求呢需统计种变化状态天全量份呢点太现实

仅浪费存储空间时业务统计点麻烦时拉链表作提现出节省空间满足需求

般数仓中通增加begin_dateen_date表示例两列start_dateend_date

1 20160820 20160820 创建 20160820 20160820
1 20160820 20160821 支付 20160821 20160821
1 20160820 20160822 完成 20160822 99991231
2 20160820 20160820 创建 20160820 20160820
2 20160820 20160821 完成 20160821 99991231
3 20160820 20160820 创建 20160820 20160821
3 20160820 20160822 支付 20160822 99991231
4 20160821 20160821 创建 20160821 20160821
4 20160821 20160822 支付 20160822 99991231
5 20160822 20160822 创建 20160822 99991231
begin_date表示该条记录生命周期开始时间end_date表示该条记录生命周期结束时间

end_date 99991231’表示该条记录目前处效状态

果查询前效记录select * from order_his where dw_end_date 99991231′

果查询20160821历史快select * from order_his where begin_date < 20160821′ and end_date > 20160821’

简单介绍拉链表更新：

假设天维度天状态天终状态

张订单表例原始数天订单状态明细

1 20160820 20160820 创建
2 20160820 20160820 创建
3 20160820 20160820 创建
1 20160820 20160821 支付
2 20160820 20160821 完成
4 20160821 20160821 创建
1 20160820 20160822 完成
3 20160820 20160822 支付
4 20160821 20160822 支付
5 20160822 20160822 创建
根拉链表希

1 20160820 20160820 创建 20160820 20160820
1 20160820 20160821 支付 20160821 20160821
1 20160820 20160822 完成 20160822 99991231
2 20160820 20160820 创建 20160820 20160820
2 20160820 20160821 完成 20160821 99991231
3 20160820 20160820 创建 20160820 20160821
3 20160820 20160822 支付 20160822 99991231
4 20160821 20160821 创建 20160821 20160821
4 20160821 20160822 支付 20160822 99991231
5 20160822 20160822 创建 20160822 99991231
出 1234订单状态统计前效状态

例hive例考虑实现性关

首先创建表

CREATE TABLE orders (
orderid INT
createtime STRING
modifiedtime STRING
status STRING
) row format delimited fields terminated by ＇＼t＇

CREATE TABLE ods_orders_inc (
orderid INT
createtime STRING
modifiedtime STRING
status STRING
) PARTITIONED BY (day STRING)
row format delimited fields terminated by ＇＼t＇

CREATE TABLE dw_orders_his (
orderid INT
createtime STRING
modifiedtime STRING
status STRING
dw_start_date STRING
dw_end_date STRING
) row format delimited fields terminated by ＇＼t＇
首先全量更新先20160820止数

初始化先20160820数初始化进

INSERT overwrite TABLE ods_orders_inc PARTITION (day ＇20160820＇)
SELECT orderidcreatetimemodifiedtimestatus
FROM orders
WHERE createtime < ＇20160821＇ and modifiedtime <＇20160821＇
刷dw中

INSERT overwrite TABLE dw_orders_his
SELECT orderidcreatetimemodifiedtimestatus
createtime AS dw_start_date
＇99991231＇ AS dw_end_date
FROM ods_orders_inc
WHERE day ＇20160820＇

结果

select * from dw_orders_his
OK
1 20160820 20160820 创建 20160820 99991231
2 20160820 20160820 创建 20160820 99991231
3 20160820 20160820 创建 20160820 99991231
剩余需进行增量更新

INSERT overwrite TABLE ods_orders_inc PARTITION (day ＇20160821＇)
SELECT orderidcreatetimemodifiedtimestatus
FROM orders
WHERE (createtime ＇20160821＇ and modifiedtime ＇20160821＇) OR modifiedtime ＇20160821＇

select * from ods_orders_inc where day＇20160821＇
OK
1 20160820 20160821 支付 20160821
2 20160820 20160821 完成 20160821
4 20160821 20160821 创建 20160821
先放增量表中然进行关联张时表中插入新表中

DROP TABLE IF EXISTS dw_orders_his_tmp
CREATE TABLE dw_orders_his_tmp AS
SELECT orderid
createtime
modifiedtime
status
dw_start_date
dw_end_date
FROM (
SELECT aorderid
acreatetime
amodifiedtime
astatus
adw_start_date
CASE WHEN borderid IS NOT NULL AND adw_end_date > ＇20160821＇ THEN ＇20160821＇ ELSE adw_end_date END AS dw_end_date
FROM dw_orders_his a
left outer join (SELECT * FROM ods_orders_inc WHERE day ＇20160821＇) b
ON (aorderid borderid)
UNION ALL
SELECT orderid
createtime
modifiedtime
status
modifiedtime AS dw_start_date
＇99991231＇ AS dw_end_date
FROM ods_orders_inc
WHERE day ＇20160821＇
) x
ORDER BY orderiddw_start_date

INSERT overwrite TABLE dw_orders_his
SELECT * FROM dw_orders_his_tmp
根面步骤20160822号数更新进结果

select * from dw_orders_his
OK
1 20160820 20160820 创建 20160820 20160820
1 20160820 20160821 支付 20160821 20160821
1 20160820 20160822 完成 20160822 99991231
2 20160820 20160820 创建 20160820 20160820
2 20160820 20160821 完成 20160821 99991231
3 20160820 20160820 创建 20160820 20160821
3 20160820 20160822 支付 20160822 99991231
4 20160821 20160821 创建 20160821 20160821
4 20160821 20160822 支付 20160822 99991231
5 20160822 20160822 创建 20160822 99991231
想数

值注意订单表中数天次状态更新应天状态天终状态天订单状态创建支付完成应拉取终状态进行拉练表更新否面数会出现异常
1 6 20160822 20160822 创建 20160822 99991231
2 6 20160822 20160822 支付 20160822 99991231
3 6 20160822 20160822 完成 20160822 99991231

Hadoop 320 完全分布式集群搭建
集群环境搭建
首先准备4台服务器(虚拟机)
设置静态ip址映射 centos7 修改静态ip设置址映射
址映射

然设置集群SSH免密登录分发脚 centos7配置集群SSH免密登录(包含群发文件脚)
果防火墙记关闭防火墙
1 systemctl stop firewalldservice
2 systemctl disable firewalldservice
载 hadoop312 分传4台服务器 roothadoop 目录
址 httparchiveapacheorgdisthadoopcommonhadoop320hadoop320targz
传 4台设备分执行解压重命名
1 cd roothadoop
2 tar zxvf roothadoophadoop320targz

安装JDK18 jdk载解压 rootjdk 重命名 jdk8
载址
httpswwworaclecomtechnetworkjavajavasedownloadsjdk8downloads2133151html

编辑 etcprofile 设置环境变量
vim etcprofile
export PATH USER LOGNAME MAIL HOSTNAME HISTSIZE HISTCONTROL 添加
1 export JAVA_HOMErootjdkjdk8
2 export JRE_HOMErootjdkjdk8jre
3 export HADOOP_HOMEroothadoophadoop320
4 PATHPATHHOMEbinJAVA_HOMEbinHADOOP_HOMEbinHADOOP_HOMEsbinrootbin
5 export LD_LIBRARY_PATHHADOOP_HOMElibnative
6 export HADOOP_CONF_DIRHADOOP_HOMEetchadoop

更新 profile 文件
source etcprofile
测试环境变量否生效
java version
hadoop version

二Hadoop配置修改
进入 hadoop 配置文件目录
cd roothadoophadoop320etchadoop
修改 hadoopenvsh 配置 jdk 路径定义集群操作户
面增加
1 export JAVA_HOMErootjdkjdk8
2
3 export HDFS_NAMENODE_USERroot
4 export HDFS_DATANODE_USERroot
5 export HDFS_SECONDARYNAMENODE_USERroot
6 export YARN_RESOURCEMANAGER_USERroot
7 export YARN_NODEMANAGER_USERroot
8
9 export HADOOP_PID_DIRroothadoopdatapids
10 export HADOOP_LOG_DIRroothadoopdatalogs

修改 coresitexml hadoop核心配置
1
2
3 fsdefaultFS
4 hdfshadoop18020
5
6
7 hadooptmpdir
8 roothadoopdatatmp
9
10
· fsdefaultFSNameNode址hadooptmpdirhadoop时目录址

修改 hdfssitexml hadoop 节点配置
1
2
3 dfsnamenodehttpaddress
4 hadoop19870
5
6
7 dfsnamenodesecondaryhttpaddress
8 hadoop250090
9
10
11 dfsreplication
12 2
13
14
15 dfsnamenodenamedir
16 fileroothadoopdatahdfsname
17
18
19 dfsdatanodedatadir
20 fileroothadoopdatahdfsdata
21
22
23
· dfsreplication 副数
· dfsnamenodesecondaryhttpaddress指定secondaryNameNodehttp访问址端口号
· 里 hadoop2 设置 SecondaryNameNode服务器

修改 workers 告知 hadoop hdfsDataNode节点
1 hadoop2
2 hadoop3
3 hadoop4

修改 yarnsitexml 配置yarn服务
1
2
3 yarnnodemanagerauxservices
4 mapreduce_shuffle
5
6
7 yarnnodemanagerlocalizeraddress
8 00008140
9
10
11 yarnresourcemanagerhostname
12 hadoop1
13
14
15 yarnresourcemanagerwebappaddress
16 hadoop18088
17
18
19 yarnlogaggregationenable
20 true
21
22
23 yarnlogaggregationretainseconds
24 604800
25
26
27 yarnlogserverurl
28 httphadoop419888jobhistorylogs
29
30
· yarnresourcemanagerwebappaddress 配置 resourcemanager 服务器址端口
· yarnresourcemanagerhostname 指定 resourcemanager 服务器
· yarnlogaggregationenable 配置否启日志聚集功
· yarnlogaggregationretainseconds 配置聚集日志HDFS保存长时间
· yarnlogserverurl 配置yarn日志服务器址
修改mapredsitexml 文件
1
2
3 mapreduceframeworkname
4 yarn
5
6
7 yarnappmapreduceamenv
8 HADOOP_MAPRED_HOMEroothadoophadoop320
9
10
11 mapreducemapenv
12 HADOOP_MAPRED_HOMEroothadoophadoop320
13
14
15 mapreducereduceenv
16 HADOOP_MAPRED_HOMEroothadoophadoop320
17
18
19 mapreducejobhistoryaddress
20 hadoop410020
21
22
23 mapreducejobhistorywebappaddress
24 hadoop419888
25
26
· yarnappmapreduceamenv mapreducemapenv mapreducereduceenv
· 三mapreduce指定hadoop目录果配置会出现运行mapreduce找main方法等错误
· mapreducejobhistoryaddress 配置务历史服务器址
· mapreducejobhistorywebappaddress 配置历史服务器web访问址

修改配置文件分发节点三台服务器
前目录执行xsync分发脚
1 xsync hadoopenvsh
2 xsync coresitexml
3 xsync hdfssitexml
4 xsync workers
5 xsync yarnsitexml
6 xsync mapredsitexml

配置完成
三Hadoop服务启动
hadoop1节点执行namenode初始格式化命令 (仅第次启动需执行)
hdfs namenode format

执行成功生成目录
cd roothadoopdatadfsname

生成集群唯id说明执行成功
hadoop1执行命令
1 startdfssh
2 startyarnsh
者执行
startallsh
hadoop4执行命令启动job历史服务
mapred daemon start historyserver
执行完成分4台设备jps查进程

启动成功
· hadoop1  NameNode ResourceManager 节点
· hadoop2  SecondaryNameNode DataNode NodeManager 节点
· hadoop3 DataNode NodeManager 节点
· hadoop4 DataNode NodeManager JobHistoryServer 节点
查HDFS Web界面 httphadoop19870

查 YARN web界面  httphadoop18088

hadoop搭建完成
四运行WordCount
首先 root 目录创建txt文件
vim roottesttxt
容
1 hadoop 1
2 hadoop 2
3 hadoop 3
4 hadoop 4
5 hadoop 5
6 hadoop 6
7 hadoop 7
8 hadoop 8
9 hadoop 9
10 hadoop 10
testtxt文件传 hdfs 执行命令
1 hdfs dfs mkdir userroot
2 hdfs dfs put roottesttxt userroot
找官方带案例jar
cd roothadoophadoop320sharehadoopmapreduce

运行jar执行MapReduce WordCount案例
hadoop jar hadoopmapreduceexamples320jar wordcount userroottesttxt rootoutput
· wordcount 第路径文件路径
· 第二路径结果输出路径 (必须存)

执行成功查执行结果
hdfs dfs lsr rootoutput

success执行成功
partr00000 m mapper 输出 r reduce 输出 00000 job 务编号整文件结果输出文件
hdfs dfs cat rootoutputpartr00000

文件中 hadoop 词出现 10次结果正确

Linux台载MySQL
1 官网载安装包
载链接：点击开链接
httpsdevmysqlcomdownloadsmysql

果系统32位选择第64位选择第二
wget 载
wget httpsdevmysqlcomgetDownloadsMySQL80mysql8011linuxglibc212i686targz
解压文件
tar zxvf mysql8011linuxglibc212i686targz
2 移动压缩包usrlocal目录重命名文件
mv rootmysql8011linuxglibc212i686 usrlocalmysql
3MySQL根目录新建文件夹data存放数
mkdir data
4创建 mysql 户组 mysql 户
1 groupadd mysql
2
3 useradd g mysql mysql

5改变 mysql 目录权限
1 chown R mysqlmysql usrlocalmysql
2
3 者
4
5 chown R mysql
6
7 chgrp R mysql
注意点

6初始化数库
创建mysql_install_db安装文件
1 mkdir mysql_install_db
2 chmod 777 mysql_install_db
初始化
binmysqld initialize usermysql basedirusrlocalmysql datadirusrlocalmysqldata 初始化数库
者
usrlocalmysqlbinmysqld initialize usermysql
1 usrlocalmysqlbinmysqld initialize usermysql
2
3 usrlocalmysqlbinmysqld (mysqld 8011) initializing of server in progress as process 5826
4
5 [Server] A temporary password is generated for root@localhost twiTlsi<0O
6
7 usrlocalmysqlbinmysqld (mysqld 8011) initializing of server has completed
记录时密码：
twiTlsi<0O

里遇问题没libnumaso1

zsh command not found mysqld
binmysqld initialize
binmysqld error while loading shared libraries libnumaso1 cannot open shared object file No such file or directory
20180429 170630 [WARNING] mysql_install_db is deprecated Please consider switching to mysqld initialize
20180429 170630 [ERROR] Can＇t locate the language directory
需安装 libnuma
1 yum install libnuma
2
3 yum y install numactl
4
5 yum install libaio1 libaiodev
安装文件

7mysql配置
cp usrlocalmysqlsupportfilesmysqlserver etcinitdmysqld
修改mycnf文件
vim etcmycnf
1
2 [mysqld]
3 basedir usrlocalmysql
4 datadir usrlocalmysqldata
5 socket usrlocalmysqlmysqlsock
6 charactersetserverutf8
7 port 3306
8 sql_modeNO_ENGINE_SUBSTITUTIONSTRICT_TRANS_TABLES
9 [client]
10 socket usrlocalmysqlmysqlsock
11 defaultcharactersetutf8
esc保存
wq 退出

8建立MySQL服务
cp a supportfilesmysqlserver etcinitdmysqld
1 cp mysqlserver etcinitdmysql
2 chmod +x etcinitdmysql
添加系统服务
chkconfig add mysql
cp a supportfilesmysqlserver etcinitdmysqld
chmod +x etcrcdinitdmysqld
chkconfig add mysqld
检查服务否生效
chkconfig list mysqld
9 配置全局环境变量
编辑 etcprofile 文件
# vi etcprofile
profile 文件底部添加两行配置保存退出
export PATHPATHusrlocalmysqlbinusrlocalmysqllib
export PATH
设置环境变量立生效
source etcprofile
10启动MySQL服务
service mysql start
查初始密码
cat rootmysql_secret
11登录MySQL
mysql uroot p密码
修改密码：
SET PASSWORD FOR ＇root＇@localhostPASSWORD(＇123456＇) #应换成密码

12设置远程登录
mysql>use mysql
mysql>update user set host＇＇ where user＇root＇ limit 1
刷新权限
mysql>flush privileges
然检查3306端口否开放
netstat nupl|grep 3306
开放3306端口
firewall cmd permanent addprot3306tcp
重启防火墙
firewall cmd reload

yum安装MySQL
安装环境：AliyunLinux（阿里linux系统64位）
cat etcosrelease

getconf LONG_BIT

首先系统中没带mysql东西先删掉
查：

find name mysql
删：

rm rf 边查找路径路径空格隔开
#者边条命令

find namemysql|xargs rm rf

开始安装
rpm Uvh httpsrepomysqlcommysql57communityreleaseel711noarchrpm

yum enablerepomysql80community install mysqlcommunityserver

步开始询问选择概意思：

总371M否载？
输入y然回车

概意思文件中检索密钥MySQL导入GPG问否OK？（英文谅解）
输入y然回车

Complete 完成
查mysql状态：
service mysqld start

接需查mysql创建默认密码首次登陆配置mysql时需
grep A temporary password varlogmysqldlog

mysql默认密码开始配置mysql
mysql_secure_installation

登陆数库：mysql u root p

功告成咯
需提醒阿里云版系统防火墙默认关闭设置果需外连接数库话记检查阿里云服务器安全组里否开放数库默认端口3306
然进入mysql库中修改update user set host＇＇ where user＇root＇
sqlyog等工具连接数库
坑：
sqlyog连接数库时出现错误提示：Authentication plugin caching_sha2_password’ cannot be loaded
客户端支持caching_sha2_password种密码加密方式
需修改密码老版密码验证方式
登陆数库进入mysql库
update user set host＇＇ where user＇root＇
重启：service mysqld restart
ALTER USER ＇root＇@＇＇ IDENTIFIED WITH mysql_native_password BY ＇新密码＇
重启：service mysqld restart

里 Abc123456a 新密码

修改退出sqlyog连接试试？

连接成功

Hive环境搭建
前提：
1 安装Hive前求先预装：
2 安装JDK 8
3 安装Hadoop277
4 安装MySQL
安装
1 载hive解压缩户目录：
1 tar xzvf apachehive236bintargz
2 改名：
3 mv apachehive236bin hive
2 设置环境变量：
二配置理
首先进入conf目录带template缀文件移缀
中hivedefaultxml移缀需修改名hivesitexml

1 通方法Hive进行配置：
11 修改hiveenvsh
1 cp hiveenvshtemplate hiveenvsh
2 Hive Hadoop 需 hiveenvsh 文件中指定 Hadoop 安装路径：
3
4 vim hiveenvsh
5
6 开配置文件中添加行：
7
8 export JAVA_HOMEusrlocalhadoopjdk180_221
9 export HADOOP_HOMEusrlocalhadoophadoop277
10 export HADOOP_CONF_DIRHADOOP_HOMEetchadoop
11 export HIVE_HOMEusrlocalhive
12 export HIVE_CONF_DIRHIVE_HOMEconf
13 export HIVE_AUX_JARS_PATHHIVE_HOMElib
12 修改hivelog4j2properties配置hivelog
1 cp hivelog4j2propertiestemplate hivelog4j2properties
2
3 vim confhivelog4j2properties
4
5 配置面参数(果没logs目录hive根目录创建)：
6
7 propertyhivelogdirusrlocalhivelogs
13 usrlocalhive215新建tmp目录tmp新建hive目录
1 cd usrlocalhive
2 mkdir tmp
3 mkdir tmphive
14 修改hivesitexml
1 cp hivedefaultxmltemplate hivesitexml
2
3 hivesitexml文件中：
4
5 {systemjavaiotmpdir}换成homehduserhivetmp
6
7 {systemusername}换1921688101 节点名
2) hivesitexml 中配置 MySQL 数库连接信息：
面配置信息需改写出需文件弄外部ctrl+f进行搜索应里数然进行修改
1
2
3
4 < 设置面属性 >
5
6 hiveexecscratchdir
7 tmphive
8
9
10
11 hiveexeclocalscratchdir
12 usrlocalhivetmphive
13 Local scratch space for Hive jobs
14
15
16
17 hivedownloadedresourcesdir
18 usrlocalhivetmp{hivesessionid}_resources
19 Temporary local directory for added resources in the remote file system
20
21
22
23 hivequeryloglocation
24 usrlocalhivetmphive
25 Location of Hive run time structured log file
26
27
28
29 hiveauxjarspath
30 usrlocalhivelibusrlocalhivejdbc
31 These JAR file are available to all users for all jobs
32
33
34 hivemetastorewarehousedir
35 hdfs 19216881019000userhivewarehouse
36 相fsdefaultname关目录理表存储位置
37
38
39 <配置Hive Metastore>
40
41 javaxjdooptionConnectionURL
42 jdbcmysql 19216881013306hivecreateDatabaseIfNotExisttrue&characterEncodingUTF8
43
44
45
46 javaxjdooptionConnectionDriverName
47 commysqljdbcDriver 高版驱动需改成commysqlcjjdbcDriver
48
49
50
51 javaxjdooptionConnectionUserName
52 root
53
54
55
56 javaxjdooptionConnectionPassword
57 123 里mysql密码
58
59
60 <配置hiveserver2机(里配置ip址便Windows连接)>
61
62 hiveserver2thriftbindhost
63 1921688101
64 Bind host on which to run the HiveServer2 Thrift service
65
66
67 <配置beeline远程客户端连接时户名密码户名应hadoop配置文件coresitexml中配置>
68
69 hiveserver2thriftclientuser
70 1921688101
71 Username to use against thrift client default is ＇anonymous＇
72
73
74
75 hiveserver2thriftclientpassword
76 123 里机户密码
77 Password to use against thrift client default is ＇anonymous＇
78
79
80 < 配置面两属性配置 hive 2x web ui >
81
82 hiveserver2webuihost
83 1921688101
84
85 < 重启HiveServer2访问http172162121710002 >
86
14 配置Hive Metastore
1 默认情况 Hive元数保存嵌derby数库里般情况生产环境MySQL存放Hive元数
2 mysqlconnectorjavaxxxjar 放入 HIVE_HOMElib （mysql jdbc驱动程序）
里注意mysql版定mysqlconnectorjavaxxxjar版低然会报错兼容（里重时卡久问题时候问题）
三运行
1 运行Hive CLI
命令行运行hive命令时必须保证HDFS已启动startdfssh启动HDFS
(特说明： Hive 21 版开始第次运行hive前需先运行schematool命令执行初始化操作)
1 果MySQL数库：
先启动mysql服务器：执行
systemctl enable mysqldservice
· 执行初始化操作
schematool initSchema dbType mysql
执行成功查MySQL中元数库hive否已创建成功
2 果derby数库：
schematool initSchema dbType derby
进入hive命令行：
hive
（hive service metastore &）
show tables 显示表
1 hive> show tables
2
3 退出hive
4 hive> quit
创建数库然创建表
1 hive> drop table chun
2 OK
3 Time taken 1125 seconds
4 hive> create database chun
5 OK
6 Time taken 0099 seconds
7 hive> use chun
8 OK
9 Time taken 0024 seconds
·
面进入HDFS web端查Hive仓库
浏览器输入：19216822813850070刚创建表

Apache Mahout环境搭建
1载解压Mahout
httparchiveapacheorgdistmahout
tar zxvf mahoutdistribution09targz

2配置环境变量
# set mahout environment
export MAHOUT_HOMEmntjediaelmahoutmahoutdistribution09
export MAHOUT_CONF_DIRMAHOUT_HOMEconf
export PATHMAHOUT_HOMEconfMAHOUT_HOMEbinPATH

3安装mahout
[jediael@master mahoutdistribution09] pwd
mntjediaelmahoutmahoutdistribution09
[jediael@master mahoutdistribution09] mvn install

4验证Mahout否安装成功
    执行命令mahout列出算法成功：
[jediael@master mahoutdistribution09] mahout
Running on hadoop using mntjediaelhadoop121binhadoop and HADOOP_CONF_DIR
MAHOUTJOB mntjediaelmahoutmahoutdistribution09examplestargetmahoutexamples09jobjar
An example program must be given as the first argument
Valid program names are
arffvector Generate Vectors from an ARFF file or directory
baumwelch BaumWelch algorithm for unsupervised HMM training
canopy Canopy clustering
cat Print a file or resource as the logistic regression models would see it
cleansvd Cleanup and verification of SVD output
clusterdump Dump cluster output to text
clusterpp Groups Clustering Output In Clusters
cmdump Dump confusion matrix in HTML or text formats
concatmatrices Concatenates 2 matrices of same cardinality into a single matrix
cvb LDA via Collapsed Variation Bayes (0th deriv approx)
cvb0_local LDA via Collapsed Variation Bayes in memory locally
evaluateFactorization compute RMSE and MAE of a rating matrix factorization against probes
fkmeans Fuzzy Kmeans clustering
hmmpredict Generate random sequence of observations by given HMM
itemsimilarity Compute the itemitemsimilarities for itembased collaborative filtering
kmeans Kmeans clustering
lucenevector Generate Vectors from a Lucene index
lucene2seq Generate Text SequenceFiles from a Lucene index
matrixdump Dump matrix in CSV format
matrixmult Take the product of two matrices
parallelALS ALSWR factorization of a rating matrix
qualcluster Runs clustering experiments and summarizes results in a CSV
recommendfactorized Compute recommendations using the factorization of a rating matrix
recommenditembased Compute recommendations using itembased collaborative filtering
regexconverter Convert text files on a per line basis based on regular expressions
resplit Splits a set of SequenceFiles into a number of equal splits
rowid Map SequenceFile to {SequenceFile SequenceFile}
rowsimilarity Compute the pairwise similarities of the rows of a matrix
runAdaptiveLogistic Score new production data using a probably trained and validated AdaptivelogisticRegression model
runlogistic Run a logistic regression model against CSV data
seq2encoded Encoded Sparse Vector generation from Text sequence files
seq2sparse Sparse Vector generation from Text sequence files
seqdirectory Generate sequence files (of Text) from a directory
seqdumper Generic Sequence File dumper
seqmailarchives Creates SequenceFile from a directory containing gzipped mail archives
seqwiki Wikipedia xml dump to sequence file
spectralkmeans Spectral kmeans clustering
split Split Input data into test and train sets
splitDataset split a rating dataset into training and probe parts
ssvd Stochastic SVD
streamingkmeans Streaming kmeans clustering
svd Lanczos Singular Value Decomposition
testnb Test the Vectorbased Bayes classifier
trainAdaptiveLogistic Train an AdaptivelogisticRegression model
trainlogistic Train a logistic regression using stochastic gradient descent
trainnb Train the Vectorbased Bayes classifier
transpose Take the transpose of a matrix
validateAdaptiveLogistic Validate an AdaptivelogisticRegression model against holdout data set
vecdist Compute the distances between a set of Vectors (or Cluster or Canopy they must fit in memory) and a list of Vectors
vectordump Dump vectors from a sequence file to text
viterbi Viterbi decoding of hidden states from given output states sequence

二简单示例验证mahout
1启动Hadoop
2载测试数
           httparchiveicsuciedumldatabasessynthetic_control链接中synthetic_controldata
者百度容易找示例数
3传测试数
hadoop fs put synthetic_controldata testdata
4 Mahout中kmeans聚类算法执行命令：
mahout core orgapachemahoutclusteringsyntheticcontrolkmeansJob
花费9分钟左右完成聚类
5查聚类结果
    执行hadoop fs ls userrootoutput查聚类结果

[jediael@master mahoutdistribution09] hadoop fs ls output
Found 15 items
rwrr 2 jediael supergroup 194 20150307 1507 userjediaeloutput_policy
drwxrxrx jediael supergroup 0 20150307 1507 userjediaeloutputclusteredPoints
drwxrxrx jediael supergroup 0 20150307 1502 userjediaeloutputclusters0
drwxrxrx jediael supergroup 0 20150307 1502 userjediaeloutputclusters1
drwxrxrx jediael supergroup 0 20150307 1507 userjediaeloutputclusters10final
drwxrxrx jediael supergroup 0 20150307 1503 userjediaeloutputclusters2
drwxrxrx jediael supergroup 0 20150307 1503 userjediaeloutputclusters3
drwxrxrx jediael supergroup 0 20150307 1504 userjediaeloutputclusters4
drwxrxrx jediael supergroup 0 20150307 1504 userjediaeloutputclusters5
drwxrxrx jediael supergroup 0 20150307 1505 userjediaeloutputclusters6
drwxrxrx jediael supergroup 0 20150307 1505 userjediaeloutputclusters7
drwxrxrx jediael supergroup 0 20150307 1506 userjediaeloutputclusters8
drwxrxrx jediael supergroup 0 20150307 1507 userjediaeloutputclusters9
drwxrxrx jediael supergroup 0 20150307 1502 userjediaeloutputdata
drwxrxrx jediael supergroup 0 20150307 1502 userjediaeloutputrandomseeds

PySpark环境搭建
Linux台载Python 38
1赖安装
1 yum y install zlibdevel bzip2devel openssldevel ncursesdevel sqlitedevel readlinedevel tkdevel gdbmdevel db4devel libpcapdevel xzdevel libffidevel
2
3 yum y install gcc
2载安装包
载安装包：httpswwwpythonorgftppython380Python380tgz
放Linux操作系统中
3解压
1 # 解压
2 tar zxvf Python380tgz C <目标文件夹选>
进入解压文件目录

4安装
1 configure prefixusrlocalpython3
2 make && make install
5添加软链接
1 ln s usrlocalpython3binpython38 usrbinpython3
2 ln s usrlocalpython3binpip38 usrbinpip3
6测试
1 python3 V
2 pip3 V

Linux升级安装python38配置pipyum
查版
安装前查否已安装python里带python275版需删情况安装python381版
python V
二安装Python381
官网载址：httpswwwpythonorgdownloadssource

1 # 解压
2 tar zxf Python381tgz
3 # 安装赖包
4 yum install zlibdevel bzip2devel openssldevel ncursesdevel sqlitedevel readlinedevel tkdevel gcc libffidevel
5 # 进入python目录
6 cd Python381
7 # 编译
8 configure prefixusrlocalpython3
9 #安装
10 make && make install
系统默认python备份
里前带python275版避免文件重名直接名字改成python275
直接Xftp直接修改命令
mv usrbinpython usrbinpython275
创建新软连接
软连接相windows新建快捷方式方便Linux需先找文件直接命令
快捷方式：Windows提供种快速启动程序开文件文件夹方法应程序快速连接
1 ln s usrlocalpython3binpython38 usrbinpython
2
3 ln s usrlocalpython3binpython38 usrbinpython3
输入面两命令python  python3命令指定 python38
面命令时候直接复制时出现点问题：
ln invalid option ＇＇
Try ＇ln help＇ for more information
果出现问题话手动敲遍面软连接命令
查python版安装成功显示：Python 381
1 python V
2
3 python3 V
三修改yum配置
升级python38yum命令会运行需修改yum应头
yumurlgrabberextdown两文件 #usrbinpython 改 #usrbinpython27
vi usrbinyum
vi usrlibexecurlgrabberextdown

四配置pip3
安装完 python381 pip install 载插件会动载 python27 带 pip 包里
pip软连接 python27 里先前 python27 版pip修改成 python38 版
备份27版软连接
mv usrbinpip usrbinpip275
配置pip3软连接 pip3python安装路径 bin 目录
1 ln s usrlocalpython3binpip3 usrbinpip
2
3 ln s usrlocalpython3binpip3 usrbinpip3
查pip版
1 pip V
2
3 pip3 V
五关yum删重新安装
1 删yum
rpm qa | grep yum | xargs rpm e nodeps
2 查Linux系统版
cat etcredhatrelease
3 查Linux核版
file binls
4 安装yum
接需载安装具体址载路径：httpmirrors163com
次执行3命令centos78网站址找指定载
1 rpm ivh nodeps httpmirrors163comcentos782003osx86_64Packagesyummetadataparser11410el7x86_64rpm
2
3 rpm ivh nodeps httpmirrors163comcentos782003osx86_64Packagesyumpluginfastestmirror113153el7noarchrpm
4
5 rpm ivh nodeps httpmirrors163comcentos782003osx86_64Packagesyum343167el7centosnoarchrpm
果linux安装python 3X版需改yum文件中配置具体见该篇文章第三部分修改yum配置
Linux安装Python 38环境卸载旧Python
前提条件
首先连接网络（会转>linux虚拟机连接网络）搭建网络yum源
cd etcyumreposd
rm rf *
wget httpmirrors163comhelpCentOS7Base163repo
yum clean all
yum makecache
· 1
· 2
· 3
· 4
· 5
安装环境
yum install gcc patch libffidevel pythondevel zlibdevel bzip2devel openssldevel ncursesdevel sqlitedevel readlinedevel tkdevel gdbmdevel db4devel libpcapdevel xzdevel y
· 1
载Python 38源代码
Windows载址：python38
wget命令linux中载
wget httpswwwpythonorgftppython380Python380a2tgz
· 1
载速度真令心寒分享百度云盘中直接载
链接：httpspanbaiducoms1O5W8G66nKoFVphheedNAfQ
提取码：ysem
安装Python 38
tar包放入linux中然执行操作
tar zxf Python380a2tgz
cd Python380a2
configure prefixusrlocalpython_38
make j 4 make install
· 1
· 2
· 3
· 4
配置
配置PATH变量
ln s usrlocalpython_38bin* usrbin
· 1

[root@test2 ]# python38
Python 380a2 (default Mar 29 2020 145852)
[GCC 485 20150623 (Red Hat 48539)] on linux
Type help copyright credits or license for more information
>>> print(＇hello＇)
hello
>>> #Ctrl+d退出
· 1
· 2
· 3
· 4
· 5
· 6
· 7
想手动操作shell脚键安装转>CentOS7中shell脚安装python38环境载
卸载原Python环境
CentOS7中默认安装python27环境需卸载卸载导致系统崩溃请谨慎处理
rpm qa|grep python|xargs rpm ev allmatches nodeps
whereis python |xargs rm frv
whereis python

安装新版Python 2713Python 362（Python 2Python 3存修改默认版Python 362）
准备工作：
1 安装wget命令（线载安装包命令）
　　yum y install wget
2 准备编译环境
1 　　yum groupinstall ＇Development Tools＇
2 　　yum install zlibdevel bzip2devel openssldevel ncursesdevel

开始安装：
1 进入载目录：
　　cd usrlocalsrc

2 载安装新版python2：
1 　　wget httpswwwpythonorgftppython2713Python2713tgz
2 　　tar zxvf Python2713tgz
3 　　cd Python2713
4 　　configure
5 　　make all
6 　　make install
7 　　make clean
8 　　make distclean
9 　　rm rf usrbinpython
10 　　rm rf usrbinpython2
11 　　rm rf usrbinpython27
12 　　ln s usrlocalbinpython27 usrbinpython
13 　　ln s usrlocalbinpython27 usrbinpython2
14 　　ln s usrlocalbinpython27 usrbinpython27
15 　　usrbinpython V
16 　　usrbinpython2 V
17 　　usrbinpython27 V
18 　　rm rf usrlocalbinpython
19 　　rm rf usrlocalbinpython2
20 　　ln s usrlocalbinpython27 usrlocalbinpython
21 　　ln s usrlocalbinpython27 usrlocalbinpython2
22 　　python V
23 　　python2 V
24 　　python27 V

3 载安装新版python3：
1 　　wget httpswwwpythonorgftppython362Python362tgz
2 　　tar zxvf Python362tgz
3 　　cd Python362
4 　　configure
5 　　make all
6 　　make install
7 　　make clean
8 　　make distclean
9 　　rm rf usrbinpython
10 　　rm rf usrbinpython3
11 　　rm rf usrbinpython36
12 　　ln s usrlocalbinpython36 usrbinpython
13 　　ln s usrlocalbinpython36 usrbinpython3
14 　　ln s usrlocalbinpython36 usrbinpython36
15 　　usrbinpython V
16 　　usrbinpython3 V
17 　　usrbinpython36 V
18 　　rm rf usrlocalbinpython
19 　　rm rf usrlocalbinpython3
20 　　ln s usrlocalbinpython36 usrlocalbinpython
21 　　ln s usrlocalbinpython36 usrlocalbinpython3
22 　　python V
23 　　python3 V
24 　　python36 V
安装pip
1 curl httpsbootstrappypaiogetpippy o getpippy # 载安装脚
2 sudo python getpippy # 运行安装脚
安装django
1 su root
2 pip install django1106

附加安装sqlite38方法
1 1 wget httpwwwsqliteorg2015sqliteautoconf3081101targz
2
3 2tar xvzf sqliteautoconf3081101targz

Linux安装Apache Spark 310详细步骤
Linux安装spark 前提部署Hadoop安装Scala
应版
名称
版
JDK
18271
Hadoop
260
Scala
2110
Apache Spark
310
第步载jdk 8u271 for linux x64
载httpswwworaclecomjavatechnologiesjavasejavasejdk8downloadshtml
wget nocookies nocheckcertificate header Cookie gpw_e24http3A2F2Fwwworaclecom2F oraclelicenseacceptsecurebackupcookie httpsdownloadoraclecomotnjavajdk8u271b0961ae65e088624f5aaa0b1d2d801acb16jdk8u271linuxx64targzAuthParam1610434774_54f5ca4ffe47aeb4b53c758f1306d437

载 httpssparkapacheorgdownloadshtml者命令行wget httpsmirrorsocfberkeleyeduapachesparkspark301spark301binhadoop27tgz

第二步解压
tar zxvf spark220binhadoop26tgz
第三步配置环境变量
vi etcprofile
#SPARK_HOME
export SPARK_HOMEhomehadoopspark220binhadoop26
export PATHSPARK_HOMEbinPATH
第四步spark配置
sparkenvsh
JAVA_HOMEhomehadoopjdk180_144
SCALA_HOMEhomehadoopscala2110
HADOOP_HOMEhomehadoophadoop260
HADOOP_CONF_DIRhomehadoophadoop260etchadoop
SPARK_MASTER_IPltt1bgcn
SPARK_MASTER_PORT7077
SPARK_MASTER_WEBUI_PORT8080
SPARK_WORKER_CORES1
SPARK_WORKER_MEMORY2g #spark里许存方默认1g 2g 里设置1g
SPARK_WORKER_PORT7078
SPARK_WORKER_WEBUI_PORT8081
SPARK_WORKER_INSTANCES1
sparkdefaultsconf
sparkmaster sparkltt1bgcn7077
slaves
ltt3bgcn
ltt4bgcn
ltt5bgcn

果整合hivehivemysql数库话需mysql数库连接驱动jmysqlconnectorjava517binjar放SPARK_HOMEjars目录

第五步spark220binhadoop26 分发节点启动
[hadoop@ltt1 sbin] startallsh
starting orgapachesparkdeploymasterMaster logging to homehadoopspark220binhadoop26logssparkhadooporgapachesparkdeploymasterMaster1ltt1bgcnout
ltt5bgcn starting orgapachesparkdeployworkerWorker logging to homehadoopspark220binhadoop26logssparkhadooporgapachesparkdeployworkerWorker1ltt5bgcnout
ltt4bgcn starting orgapachesparkdeployworkerWorker logging to homehadoopspark220binhadoop26logssparkhadooporgapachesparkdeployworkerWorker1ltt4bgcnout
ltt3bgcn starting orgapachesparkdeployworkerWorker logging to homehadoopspark220binhadoop26logssparkhadooporgapachesparkdeployworkerWorker1ltt3bgcnout
查进程
master节点
[hadoop@ltt1 sbin] jps
1346 NameNode
1539 JournalNode
1812 ResourceManager
1222 QuorumPeerMain
1706 DFSZKFailoverController
2588 Master
2655 Jps
worker节点
[hadoop@ltt5 ~] jps
1299 NodeManager
1655 Worker
1720 Jps
1192 DataNode
进入SparkWeb理页面： httpltt1bgcn8080

spark安装完成
1 解压缩安装包 tar xvf jdk8u271linuxx64targz

1 进入解压缩jdk文件中pwd查前工作路径然修改文件vi ~bash_profile

3 ~bash_profile 文件末尾加jdk环境变量容：
4刚修改文件生效java version查否配置成功

Spark安装配置
1 解压spark安装包tar xvf spark243binhadoop27tgz

2 进入解压文件中进入conf目录查配置文件

3修改配置文件sparkenvsh注意文件默认存里sparkenvshtemplate复制份命名新文件spaekenvsh

4查前JAVA_HOME路径步中

5修改文件sparkenvsh文件末尾添加容：

6回spark目录中找sbin目录然启动spark命令sbinstartallsh

7jps查否启动成功

8spark根目录examplesjars目录jar文件里面存放例子

9里jar包进行测试求圆周率

10回spark目录运行命令里面100设置值选择更值进行测试会更精确

结果显示：

11创建两目录inputoutput作文件输入输出目录

12输入目录中创建datatxt文件容

13启动sparkshell交互式工具黄框标记日志表示变量sc操作Spark context

14spark中scala语言统计单词出现次数
sctextFile读取文件split( )空格分隔字符 map((_1))单词计数里元祖
reduceByKey相进行累加

Spark集群安装设置
spark100新版20140530正式发布啦新spark版带新特性提供更API支持spark100增加Spark SQL组件增强标准库（MLstreamingGraphX）JAVAPython语言支持
面首先进行spark100集群安装里两台服务器台作masternamenode机台作slavedatanode机增加更slave需重复slave部分容：
系统版：
· master：Ubuntu 1204
· slave：Ubuntu 1204
· hadoop：hadoop 220
· spark：spark 100
1 安装JDKhadoop集群
安装程参见里httpwwwcnblogscomtecvegetablesp3778358html

2 载安装Scala
· scala载址httpwwwscalalangorgdownload2111html里载新版scala2111版
· 解压scala放usrlib目录
　　　　tar xzvf scala2111tgz
　　　　mv scala2111 usrlib
· 配置scala环境变量：sudo vi etcprofile
　　　文件末尾添加scala路径
　　　
　　　输入 source etcprofile 路径生效
· 测试scala：scala version   #出现scala版信息说明安装成功
PS：scala需slave节点配置
3 载安装spark
· spark100载址httpsparkapacheorgdownloadshtml解压spark放homehadoop
　　tar xzvf spark100binhadoop2tgz
· 配置spark环境变量：sudo vi etcprofile
　　文件末尾添加spark路径

　　输入 source etcprofile  路径生效
· 配置confsparkenvsh文件
　　没该文件 sparkenvshtemplate 文件重命名文件中添加scalajavahadoop路径master ip等信息
　　mv sparkenvshtemplate sparkenvsh
　　vi sparkenvsh
　　
　　
　　
· confslaves中添加slave节点hostname行：
　　vi slaves
　　
4 slave机器安装配置spark
现master机spark文件分发slave节点注意slavemasterspark目录必须致master会登录slave执行命令认slavespark路径样
scp r spark100binhadoop2 hadoop@slavehomehadoop
5启动spark集群
master机执行命令：
cd ~spark100binhadoop2sbin
startallsh
检测进程否启动：输入 jps

配置完成
6 面体验spark带例子
binrunexample SparkPi

scala实现spark app
官方说明址例子统计输入文件中字母a字母b数网站提供scalajavapython三种实现里做scala吧里需安装SBT（ sbt 创建测试运行提交作业简单SBT做Scala世界Maven）
spark100木带sbt选择手动安装然选择sudo aptget install sbt方式（系统中木找sbt包手动安装咯）安装方法：
· 载：sbt载址载现新版sbt0135
· 解压sbthomehadoop目录（hadoop户名实HOME啦）
　　tar zxvf sbt0135tgz
　　cd sbtbin
　　java jar sbtlaunchjar   #进行sbt安装时间约时吧会载东东记联网哦
· 成功etcprofile中配置sbt环境变量
　　sudo vi etcprofile
　　
　　输入source etcprofile 路径生效
sbt安装完成面写简单spark app吧
· 创建目录：mkdir ~SimpleApp
· SimpleApp目录创建目录结构：

· simplesbt文件容：
name Simple Project
version 10
scalaVersion 2104
libraryDependencies + orgapachespark sparkcore 100
resolvers + Akka Repository at httprepoakkaioreleases
· SimpleAppscala文件容：
* SimpleAppscala *
import orgapachesparkSparkContext
import orgapachesparkSparkContext_
import orgapachesparkSparkConf
object SimpleApp {
def main(args Array[String]) {
val logFile YOUR_SPARK_HOMEREADMEmd Should be some file on your system
val conf new SparkConf()setAppName(Simple Application)
val sc new SparkContext(conf)
val logData sctextFile(logFile 2)cache()
val numAs logDatafilter(line > linecontains(a))count()
val numBs logDatafilter(line > linecontains(b))count()
println(Lines with a s Lines with b sformat(numAs numBs))
}
}
PS：前spark配置程中hadoop路径配置里输入路径YOUR_SPARK_HOMEXXX实际HDFS文件系统中文件存储位置hadoop配置文件coresitexml中相关（具体参见里方容易出错）需先READMEmd文件puthdfs面：

· 编译：
　　cd ~SimpleApp
　　sbt package   #包程时间会较长会出现[success]XXX
　　PS：成功会生成许文件 targetscala210simpleproject_21010jar等
· 运行：
　　sparksubmit class SimpleApp master local targetscala210simpleproject_21010jar
· 结果：

7 停止spark集群
cd ~spark100binhadoop2sbin
stopallsh

JDKhadoop安装
Ubuntu 1204Hadoop 220 集群搭建
现家起实现Ubuntu 1204Hadoop 220 集群搭建里两台服务器台作masternamenode机台作slavedatanode机增加更slave需重复slave部分容
系统版：
· master：Ubuntu 1204
· slave：Ubuntu 1204
· hadoop：hadoop 220
· 安装ssh服务：sudo aptget install ssh
· 时更新vim：sudo aptget install vim  #刚安装系统会出现vi 命令键变成AB情况
masterslave机器安装JDK环境
载jdk果安装版170_60官方载址：java载
解压jdk： tar xvf jdk7u60linuxi586targz
usrlocal新建java文件夹：mkdir usrlocaljava
解压文件移动创建java文件夹：sudo mv jdk170_60  usrlocaljava
修改etcprofile文件：sudo vi etcprofile
文件末尾添加jdk路径：

输入 source etcprofile java生效
测试java否完全安装：java version   #出现版信息说明安装成功
二修改namenode（master）子节点（slave）机器名：
sudo vi etchostname

修改需重启生效：sudo reboot
三修改namenode（master）节点映射ip
sudo vi etchosts #添加slavemaster机器名应ip

PS：masterslave分namenodedatanode机器名hostname名字
四masterslave分创建Hadoop户户组赋予sudo权限
sudo addgroup hadoop
sudo adduser ingroup hadoop hadoop   #第hadoop户组第二hadoop户名
面hadoop户赋予sudo权限：修改 etcsudoers 文件
sudo vi etcsudoers
添加hadoop ALL(ALLALL) ALL

PS：该操作需masterslave机进行
五建立ssh密码登陆环境
hadoop身份登录系统：su hadoop
生成密钥建立namenodedatanode信关系ssh生成密钥rsadsa方式默认采rsa方式：
homehadoop目录输入： sshkeygen t rsa P
确认信息回车会homehadoopssh生成文件：
id_rsapub追加authorized_keys授权文件中： cat id_rsapub >> authorized_keys

子节点生成密钥：sshkeygen t rsa P
masterauthorized_keys发送子节点：
scp ~sshauthorized_keys hadoop@slave1~ssh
面测试ssh互信： ssh hadoop@slave1
果需输入密码登录成功表示ssh互信成功建立
六安装hadoop（需配置master机slave机直接复制）
载hadoopusrlocal：载址
解压hadoop220targz：sudo tar zxf hadoop220targz
解压出文件夹重命名hadoop： sudo mv hadoop220 hadoop
hadoop文件夹属户设hadoop：sudo chown R hadoophadoop hadoop
1配置etchadoophadoopenvsh文件
sudo vi usrlocalhadoopetchadoophadoopenvsh
找export JAVA_HOME部分修改机jdk路径

2配置etchadoopcoresitexml文件
sudo vi usrlocalhadoopetchadoopcoresitexml
中间添加容：

PS：masternamenode机名etchosts文件里名字
3配置etchadoopmapredsitexml文件路径没文件mapredsitexmltemplate重命名
sudo vi usrlocalhadoopetchadoopmapredsitexml
中间添加容：

PS：masternamenode机名etchosts文件里名字
4配置hdfssitexml文件路径没文件hdfssitexmltemplate重命名
sudo vi hdfssitexml
中间添加容：

PS：slave节点间容改：
· usrlocalhadoopdatalog1usrlocalhadoopdatalog2
· usrlocalhadoopdata1usrlocalhadoopdata2
· 1中间数字表示slave节点数
5slaves文件中添加slave机名行

七slave节点分发配置文件
配置文件发送slave子节点先文件复制子节点homehadoop面（子节点hadoop户登录：su hadoop）
sudo scp etchosts hadoop@slave1homehadoop
scp r usrlocalhadoop hadoop@slave1homehadoop
PS：slave1slave子节点名slave节点应全部分发
datanode机器（slave节点）文件移动master相路径
sudo mv homehadoophosts etchosts  (子节点执行)
sudo mv homelocalhadoop usrlocal  (子节点执行)
PS：提示mv文件夹加r 参数
加入属户： sudo chown R hadoophadoop hadoop    (子节点执行)
PS：子节点datanode机器复制hadoop里面data1data2logs删掉
配置完成
PS：hadoop命令路径写入etcprofile文件样hadoophdfs命令否命令时加入binhadoop样路径：
sudo vi etcprofile

输入：source etcprofile
八运行WordCount示例
首先进入usrlocalhadoop目录重启hadoop
cd usrlocalhadoopsbin
stopallsh
cd usrlocalhadoopbin
hdfs namenode format # 格式化集群
cd usrlocalhadoopsbin
startallsh
namenode查连接情况
hdfs dfsadmin report #面机器结果

假设测试文件test1txttest2txt首先创建目录input
hadoop dfs mkdir input
测试文件传hadoop：
hadoop dfs put test1txt input
hadoop dfs put test2txt input
子节点离开安全模式否会导致法读取input文件：
hdfs dfsadmin –safemode leave
运行wordcount程序：
hadoop jar usrlocalhadoopsharehadoopmapreducehadoopmapreduceexamples220jar wordcount input output
查结果
hadoop dfs cat outputpartr00000
PS：次运行时需删output文件夹：hadoop dfs rmr output
参考资料：
httpwwwcnblogscomkinglaup3794433html
httpwwwcnblogscomtecvegetablesp3778358html
httpblogcsdnnetlaoyi_gracearticledetails6254743

Ubuntu 1404安装Hadoop240（单机模式）
Ubuntu创建hadoop组hadoop户
    增加hadoop户组时该组里增加hadoop户续涉hadoop操作时该户
1创建hadoop户组

2创建hadoop户
    sudo adduser ingroup hadoop hadoop
    回车会提示输入新UNIX密码新建户hadoop密码输入回车
    果输入密码回车会重新提示输入密码密码空
    确认信息否正确果没问题输入 Y回车
3hadoop户添加权限
     输入：sudo gedit etcsudoers
     回车开sudoers文件
hadoop户赋予root户样权限


二新增加hadoop户登录Ubuntu系统

三安装ssh
sudo aptget install opensshserver安装完成启动服务
sudo etcinitdssh start

查服务否正确启动：ps e | grep ssh

设置免密码登录生成私钥公钥
sshkeygen t rsa P

cat ~sshid_rsapub >> ~sshauthorized_keys

登录ssh
ssh localhost

退出
exit

四安装Java环境
sudo aptget install openjdk7jdk

查安装结果输入命令：java version结果表示安装成功

五安装hadoop240
    1官网载httpmirrorbiteducnapachehadoopcommon

    2安装

        解压
        sudo tar xzf hadoop240targz
        假hadoop安装usrlocal
        拷贝usrlocal文件夹hadoop
        sudo mv hadoop240 usrlocalhadoop


赋予户该文件夹读写权限
        sudo chmod 774 usrlocalhadoop


3配置
        1）配置~bashrc
配置该文件前需知道Java安装路径设置JAVA_HOME环境变量面命令行查安装路径
        updatealternatives config java
        执行结果：

完整路径
    usrlibjvmjava7openjdkamd64jrebinjava
    取前面部分 usrlibjvmjava7openjdkamd64
    配置bashrc文件
    sudo gedit ~bashrc

    该命令会开该文件编辑窗口文件末尾追加面容然保存关闭编辑窗口
#HADOOP VARIABLES START
export JAVA_HOMEusrlibjvmjava7openjdkamd64
export HADOOP_INSTALLusrlocalhadoop
export PATHPATHHADOOP_INSTALLbin
export PATHPATHHADOOP_INSTALLsbin
export HADOOP_MAPRED_HOMEHADOOP_INSTALL
export HADOOP_COMMON_HOMEHADOOP_INSTALL
export HADOOP_HDFS_HOMEHADOOP_INSTALL
export YARN_HOMEHADOOP_INSTALL
export HADOOP_COMMON_LIB_NATIVE_DIRHADOOP_INSTALLlibnative
export HADOOP_OPTSDjavalibrarypathHADOOP_INSTALLlib
#HADOOP VARIABLES END

终结果图：

执行面命添加环境变量生效：
        source ~bashrc
2）编辑usrlocalhadoopetchadoophadoopenvsh
         执行面命令开该文件编辑窗口
        sudo gedit usrlocalhadoopetchadoophadoopenvsh
找JAVA_HOME变量修改变量
        export JAVA_HOMEusrlibjvmjava7openjdkamd64
修改hadoopenvsh文件示：

六WordCount测试
单机模式安装完成面通执行hadoop带实例WordCount验证否安装成功
    usrlocalhadoop路径创建input文件夹
mkdir input

    拷贝READMEtxtinput
cp READMEtxt input
    执行WordCount
    binhadoop jar sharehadoopmapreducesourceshadoopmapreduceexamples240sourcesjar orgapachehadoopexamplesWordCount input output  执行结果：
  执行 cat output*查字符统计结果

启动Spark集群
检查进程否启动Spark集群环境否搭建成功
1）启动HDFS集群

2）检查进程否启动

3）启动Spark集群
cd homehadoopspark121binhadoop24sbin
startallsh
jps （没启动hdfs 集群spark进程情况）
jps（启动hdfs集群spark进程运行情况）
浏览器输入httpmaster8080查spark集群运行状况

6）进入Sparkbin目录启动sparkshell 控制台

7）访问 httpmaster4040Spark WebUI页面Spark集群环境搭建成功

8）运行sparkshell测试
前userinput传READMEtxt文件现Spark读取HDFS中READMEtxt文件

取HDFS文件
（连接成功例子）

连接成功例子似Hadoop26版识localhost机IP址

CountREADMEtxt文件中文字总数

滤READMEtxt文件
包括单词

通wcREADMEtxt统计4The 单词

实现Hadoop wordcount 功
首先读取readmeFile执行命令

然collect命令提交job

WebUI执行效果：

cd
binrunexample SparkPi

停止集群
sbinstopallsh

OkSpark集群环境测试结束总结步骤：
（1）cd usrlocalhadoopsbin
startallsh
（2）cd homehadoopspark121binhadoop24sbin
startallsh
（3）jps
（4）http19216801188080
（5）cd homehadoopspark121binhadoop24bin
sparkshell
（6）http19216801184040
（7）scala编程程序提交测试
（8）集群停止
Spark性优化
般开发完Spark作业该作业配置合适资源Spark资源参数基sparksubmit命令中作参数设置资源参数设置会导致没充分利集群资源作业运行会极缓慢者设置资源队列没足够资源提供进导致种异常

pyspark开发代码例子说明
运行pyspark程序终端命令模式Linux终端输入pyspark然复制粘贴代码sparksubmit命令行Hive样yarn调度运行

# *codingutf8*
from pysparksql import HiveContext SparkSession

# 初始化SparkContext时启Hive支持
# 终端命令行测试模式输出字段长度设置100字符
spark SparkSessionbuilderappName(name)config(
sparkdebugmaxToStringFields 100)enableHiveSupport()getOrCreate()
# 初始化HiveContext
hive HiveContext(sparksparkContext)
# 启SparkSQL表连接支持
sparkconfset(sparksqlcrossJoinenabled true)

# 读取parquet文件数代码
# Parquet面分析型业务列式存储格式TwitterCloudera合作开发AWS中
# parquet文件数存储AWS S3
# AWSS3作数存储服务S3 全名 Simple Storage Service简便存储服务
df1 sparkreadload(
path＇＇
format＇parquet＇ headerTrue)

# 读取CSV文件数代码
# 边CSV文件作手工交换文件标准
# 原csv格式简单数字类型数字符串存储精度保证
df2 sparkreadload(
path＇＇
format＇csv＇ headerTrue)

# 读取Hive表视图数代码
df3 hivesql(
select
*
from <数库名><表名>)

# 次表数集进行数存缓存（第条Spark优化策略）
# 样话pyspark代码次调数时候Spark会重复读取相文件数
df4 sparkreadload(
path＇＇
format＇parquet＇ headerTrue)cache()

# 刚数集命名便放入SparkSQL编写查询语句
df1createOrReplaceTempView(DF1)

df2createOrReplaceTempView(DF2)

df3createOrReplaceTempView(DF3)

df4createOrReplaceTempView(DF4)

# 创建SparkSQL数集代码
# 果数量较业务逻辑复杂话数时缓存存储服务磁盘
# 避免pyspark代码SparkSQL调里SparkSQL数集时候
# 里SparkSQL数集重复运行计算逻辑节约计算资源（第二条Spark优化策略）
df5 sparksql(
SELECT

from DF1 AS D1
LEFT JOIN DF2 AS D2
ON
LEFT JOIN DF4 AS D4
ON
WHERE
)persist()
# countAction算子会触发sparksubmit事件前persist()缓存操作刻生效
# count()操作persist()缓存操作会Action算子处程序结束处生效
df5count()
df5createOrReplaceTempView(DF5)

# 创建SparkSQL数集代码
df6 sparksql(
SELECT

from DF5 AS D5
LEFT JOIN DF3 AS D3
ON
LEFT JOIN DF4 AS D4
ON
WHERE
)

# 写入结果数集parquet文件
df6writeparquet(
path＇＇
modeoverwrite)

# 释放磁盘缓存
df5unpersist()

# sparkContext停止
sparkstop()

1Spark作业基运行原理

详细原理见图sparksubmit提交Spark作业作业会启动应Driver进程根部署模式（deploymode）Driver进程启动集群中某工作节点启动Driver进程身会根设置参数占定数量存CPU coreDriver进程做第件事情集群理器（Spark Standalone集群资源理集群般YARN作资源理集群）申请运行Spark作业需资源里资源指Executor进程YARN集群理器会根Spark作业设置资源参数工作节点启动定数量Executor进程Executor进程占定数量存CPU core
　　申请作业执行需资源Driver进程会开始调度执行编写作业代码Driver进程会编写Spark作业代码分拆stagestage执行部分代码片段stage创建批task然task分配Executor进程中执行task计算单元负责执行模样计算逻辑（编写某代码片段）task处理数已stagetask执行完毕会节点磁盘文件中写入计算中间结果然Driver会调度运行stagestagetask输入数stage输出中间结果循环复直编写代码逻辑全部执行完计算完数想结果止
　　Spark根shuffle类算子进行stage划分果代码中执行某shuffle类算子（reduceByKeyjoin等）会该算子处划分出stage界限致理解shuffle算子执行前代码会划分stageshuffle算子执行代码会划分stagestage刚开始执行时候task会stagetask节点通网络传输拉取需处理key然拉取相key编写算子函数执行聚合操作（reduceByKey()算子接收函数）程shuffle
　　代码中执行cachepersist等持久化操作时根选择持久化级task计算出数会保存Executor进程存者节点磁盘文件中
　　Executor存分三块：第块task执行编写代码时默认占Executor总存20第二块task通shuffle程拉取stagetask输出进行聚合等操作时默认占Executor总存20第三块RDD持久化时默认占Executor总存60
　　task执行速度Executor进程CPU core数量直接关系CPU core时间执行线程Executor进程分配tasktask条线程方式线程发运行果CPU core数量较充足分配task数量较合理通常说较快速高效执行完task线程
　　Spark作业基运行原理说明家结合图理解理解作业基原理进行资源参数调优基前提
2资源参数调优
解完Spark作业运行基原理资源相关参数容易理解谓Spark资源参数调优实Spark运行程中资源方通调节种参数优化资源效率提升Spark作业执行性参数Spark中资源参数参数应着作业运行原理中某部分时出调优参考值
numexecutors
　　参数说明：该参数设置Spark作业总少Executor进程执行DriverYARN集群理器申请资源时YARN集群理器会设置集群工作节点启动相应数量Executor进程参数非常重果设置话默认会启动少量Executor进程时Spark作业运行速度非常慢
　　参数调优建议：Spark作业运行般设置50~100左右Executor进程较合适设置太少太Executor进程设置太少法充分利集群资源设置太话部分队列法予充分资源
executormemory
　　参数说明：该参数设置Executor进程存Executor存时候直接决定Spark作业性常见JVM OOM异常直接关联
　　参数调优建议：Executor进程存设置4G~8G较合适参考值具体设置根部门资源队列定团队资源队列存限制少numexecutorsexecutormemory代表Spark作业申请总存量（Executor进程存总）量超队列存量外果团队里享资源队列申请总存量超资源队列总存13~12避免Spark作业占队列资源导致学作业法运行
executorcores
　　参数说明：该参数设置Executor进程CPU core数量参数决定Executor进程行执行task线程力CPU core时间执行task线程Executor进程CPU core数量越越够快速执行完分配task线程
　　参数调优建议：ExecutorCPU core数量设置2~4较合适样根部门资源队列定资源队列CPU core限制少设置Executor数量决定Executor进程分配CPU core样建议果享队列numexecutors * executorcores超队列总CPU core13~12左右较合适避免影响学作业运行
drivermemory
　　参数说明：该参数设置Driver进程存
　　参数调优建议：Driver存通常说设置者设置1G左右应该够唯需注意点果需collect算子RDD数全部拉取Driver进行处理必须确保Driver存足够否会出现OOM存溢出问题
sparkdefaultparallelism
　　参数说明：该参数设置stage默认task数量参数极重果设置会直接影响Spark作业性
　　参数调优建议：Spark作业默认task数量500~1000较合适学常犯错误设置参数时会导致Spark根底层HDFSblock数量设置task数量默认HDFS block应task通常说Spark默认设置数量偏少（十task）果task数量偏少话会导致前面设置Executor参数前功弃试想Executor进程少存CPUtask1者1090Executor进程根没task执行白白浪费资源Spark官网建议设置原设置该参数numexecutors * executorcores2~3倍较合适Executor总CPU core数量300设置1000task时充分利Spark集群资源
sparkstoragememoryFraction
　　参数说明：该参数设置RDD持久化数Executor存中占例默认06说默认Executor 60存保存持久化RDD数根选择持久化策略果存够时数会持久化者数会写入磁盘
　　参数调优建议：果Spark作业中较RDD持久化操作该参数值适提高保证持久化数够容纳存中避免存够缓存数导致数写入磁盘中降低性果Spark作业中shuffle类操作较持久化操作较少参数值适降低较合适外果发现作业频繁gc导致运行缓慢（通spark web ui观察作业gc耗时）意味着task执行户代码存够样建议调低参数值
sparkshufflememoryFraction
　　参数说明：该参数设置shuffle程中task拉取stagetask输出进行聚合操作时够Executor存例默认02说Executor默认20存进行该操作shuffle操作进行聚合时果发现存超出20限制余数会溢写磁盘文件中时会极降低性
　　参数调优建议：果Spark作业中RDD持久化操作较少shuffle操作较时建议降低持久化操作存占提高shuffle操作存占例避免shuffle程中数时存够必须溢写磁盘降低性外果发现作业频繁gc导致运行缓慢意味着task执行户代码存够样建议调低参数值
资源参数调优没固定值需学根实际情况（包括Spark作业中shuffle操作数量RDD持久化操作数量spark web ui中显示作业gc情况）时参考篇文章中出原理调优建议合理设置述参数
3资源参数参考示例
份sparksubmit命令示例家参考根实际情况进行调节：
binsparksubmit ＼
master yarncluster ＼
numexecutors 100 ＼
executormemory 6G ＼
executorcores 4 ＼
drivermemory 1G ＼
conf sparkdefaultparallelism1000 ＼
conf sparkstoragememoryFraction05 ＼
conf sparkshufflememoryFraction03 ＼
4Spark中三种Join策略
Spark通常三种Join策略方式
1 Broadcast Hash Join（BHJ）
2 Shuffle Hash Join（SHJ）
3 Sort Merge Join（SMJ）
Broadcast Hash Join
表表进行Join操作时避免shuffle操作表数分发节点表进行Join操作牺牲空间避免耗时Shuffle操作

1 表需broadcast必须sparksqlautoBroadcastJoinThreshold配置值（默认10M）者明确添加broadcast join hint
2 base tablebroadcast例left outer join中仅仅right表broadcast
3 种算法仅仅broadcast表否数传输shuffle操作成高
4 broadcast需driver果太broadcastdriver存压力
Shuffle Hash Join
broadcast策略首先收集数driver节点然分发executor节点表太时broadcastchelve会driverexecutor造成压力
Shuffle Hash Join会减少driverexeuctor压力操作步骤：
1 两张表分连接列进行重组目相连接列记录分配分区
2 两张表表分区构造成hash表表根相应记录进行映射

Sort Merge Join
面两种发现适应定表两张表足够时面方法存造成压力两张表做Hash Join时中张表必须完成加载存中
两张表时Spark SQL种新算法做Join操作做Sort Merge Join种算法会加载数然开始Hash JoinJoin前进行数排序
两张表需进行数重组保证相连接列值分区中分区数排序分区中数然相应记录进行关联
表分区Sort Merge Join会加载张表数存

5Spark 30 中 AQE新特性
年Spark SQL优化CBO成功特性
CBO会计算业务数相关统计数优化查询例行数重行数空值值等
Spark根数动选择BHJ者SMJJoin场景Costbased Join Reorder（参考前写篇文章）达优化执行计划目
统计数需预先处理会时时数进行判断某情况反会变成负面效果拉低SQL执行效率
AQE执行程中统计数动态调节执行计划解决问题
1框架
AQE言重问题什时候重新计算优化执行计划Spark务算子果道排列次行执行然shuffle者broadcast exchange会断算子排列执行称物化点（Materialization Points）Query Stages代表物化点分割片段Query Stage会产出中间结果仅该stage行stage执行完成游Query Stage执行游部分stage执行完成partitions统计数获取游未开始执行AQE提供reoptimization机会

查询开始时生成完执行计划AQE框架首先会找执行存游stages旦stage完成AQE框架会physical plan中标记完成根已完成stages提供执行数更新整logical plan基新产出统计数AQE框架会执行optimizer根系列优化规进行优化AQE框架会执行生成普通physical planoptimizer适应执行专属优化规例分区合数倾斜处理等获新优化执行计划已执行完成stages次循环接着需继续重复面步骤直整query跑完
Spark 30中AQE框架拥三特征：
· 动态折叠shuffle程中partition
· 动态选择join策略
· 动态优化存数倾斜join
接具体三特征
① 动态合shuffle partitions
处理数量级非常时shuffle通常说影响性shuffle非常耗时算子需通网络移动数分发游算子
shuffle中partition数量十分关键partition佳数量取决数数querystage会差异难确定具体数目：
· 果partition少partition数量会会导致量数落磁盘拖慢查询
· 果partitionpartition数量会少会产生额外网络开销影响Spark task scheduler拖慢查询
解决该问题开始设置相较shuffle partition数通执行程中shuffle文件数合相邻partitions
例假设执行SELECT max(i) FROM tbl GROUP BY j表tbl2partition数量非常初始shuffle partition设5分组会出现5partitions进行AQE优化会产生5tasks做聚合结果事实3partitions数量非常

然种情况AQE会生成3reduce task

② 动态切换join策略
Spark支持众join中broadcast hash join性果需广播表预估广播限制阈值应该设BHJ表估计会导致决策错误join表filter（容易表估）者join表算子（容易表估）仅仅全量扫描张表
AQE拥精确游统计数解决该问题面例子右表实际15M该场景filter滤实际参join数8M默认broadcast阈值10M应该广播

执行程中转化BHJ时甚传统shuffle优化shuffle（例shuffle读mapper基reducer）减网络开销
③ 动态优化数倾斜
数倾斜集群数分区间分布均匀导致会拉慢join场景整查询AQE根shuffle文件统计数动检测倾斜数倾斜分区散成子分区然进行join
场景Table A join Table B中Table Apartition A0数远分区

AQE会partition A0切分成2子分区独Table Bpartition B0进行join

果做优化SMJ会产生4tasks中执行时间远优化join会5taskstask执行耗时差相整查询带更性
2
设置参数sparksqladaptiveenabledtrue开启AQESpark 30中默认false满足条件：
· 非流式查询
· 包含少exchange（join聚合窗口算子）者子查询
AQE通减少静态统计数赖成功解决Spark CBO难处理trade off（生成统计数开销查询耗时）数精度问题相前具局限性CBO现显非常灵活需提前分析数
6数仓库中数优化般原
l 种数放份：两张张表中种数字段数量字段键抽出变成张独立新表（数仓库设计）
l ETL中数筛选聚合操作量前置取数源数方减少整数流数量（ETL设计）
l 张表连接时提前张表中连接数进行排表连接时间复杂度笛卡尔积减少数性提高较明显应筛选数中参连接字段添加索引便进步提高表连接性

7Spark 中宽赖窄赖
Spark中RDD高效DAG图着莫关系DAG调度中需计算程划分stage划分RDD间赖关系针转换函数RDD间赖关系分类窄赖（narrow dependency）宽赖（wide dependency 称 shuffle dependency）
概述
· 窄赖指父RDD分区子RDD分区子RDD分区通常应常数父RDD分区(O(1)数规模关)
· 相应宽赖指父RDD分区子RDD分区子RDD分区通常应父RDD分区(O(n)数规模关)
宽赖窄赖图示：

相宽赖窄赖优化利基两点：
1 宽赖应着shuffle操作需运行程中父RDD分区传入子RDD分区中中间涉节点间数传输窄赖父RDD分区会传入子RDD分区中通常节点完成转换
2 RDD分区丢失时（某节点障）spark会数进行重算
1 窄赖父RDD分区应子RDD分区样需重算子RDD分区应父RDD分区重算数利率100
2 宽赖重算父RDD分区应子RDD分区样实际父RDD 中部分数恢复丢失子RDD分区部分应子RDD未丢失分区造成余计算更般宽赖中子RDD分区通常父RDD分区极端情况父RDD分区进行重新计算
3 图示b1分区丢失需重新计算a1a2a3产生冗余计算(a1a2a3中应b2数)

详细运行原理

图中左边宽赖父RDD4号分区数划分子RDD分区（分区分区）表明shuffle程父分区数shuffle程hash分区器（定义分区器）划分子RDD例GroupByKeyreduceByKeyjoinsortByKey等操作
图右边窄赖父RDD分区数直接子RDD应分区（分区分区）例1号5号分区数进入子RDD分区程没shuffleSpark中Stage划分通shuffle划分（shuffle理解数原分区乱重组新分区）：mapfilter
总结：果父RDDPartition子RDDPartition窄赖否话宽赖
宽窄赖容错性
Spark基lineage容错性指果RDD出错父RDD重新计算果RDD仅父RDD（窄赖）种重新计算代价会非常

Spark基Checkpoint(物化)容错机制解？图中宽赖结果(历Shuffle程）昂贵Spark结果物化磁盘备面
join操作两种情况果join操作partition 仅仅已知Partition进行join时join操作窄赖情况join操作宽赖确定Partition数量赖关系窄赖出推窄赖仅包含窄赖包含固定数窄赖（说父RDD赖Partition数量会着RDD数规模改变改变）
Stage划分
名词解析
1 job rdd action 触发动作简单理解需执行 rdd action 时候会生成 job
2stage stage job 组成单位说 job 会切分成 1 1 stage然 stage 会执行序次执行
3task stage 务执行单元般说 rdd 少partition会少 task task 处理partition 数
划分规
1前推理遇宽赖断开遇窄赖前RDD加入Stage中
2Stage里面Task数量该Stage中 RDDPartition数量决定
3Stage里面务类型ResultTask前面Stage里面务类型ShuffleMapTask
4代表前Stage算子定该Stage计算步骤
总结：spark中stage划分根shuffle划分宽赖必然shuffle程说spark根宽窄赖划分stage
Spark优化
窄赖优化利逻辑RDD算子forkjoin（join非文join算子指步行务barrier）：计算fork分区算完join然forkjoinRDD算子果直接翻译物理实现济：RDD（中间结果）需物化存存储中费时费空间二join作全局barrier昂贵会慢节点拖死果子RDD分区父RDD分区窄赖实施典fusion优化两forkjoin合果连续变换算子序列窄赖 forkjoin减少量全局barrier需物化中间结果RDD极提升性Spark做流水线（pipeline）优化
Spark流水线优化：

变换算子序列碰shuffle类操作宽赖发生流水线优化终止具体实现中DAGScheduler前算子前回溯赖图碰宽赖生成stage容纳已遍历算子序列stage里安全实施流水线优化然宽赖开始继续回溯生成stage
Pipeline
Spark中pipelinepartition应partitionstage部窄赖pipeline详解
stagestage间宽赖
分布式计算程

图Sparkwordcount例子根述stage划分原job划分2stage三行分数读取计算存储程
仅代码户根体会数背行计算图中出数分布分区（理解机器）数flapMapmapreduceByKey算子RDD分区中流转（算子面说RDD进行计算函数）
图更高角度：

Spark运行架构Driver（理解master）Executor（理解workerslave)组成Driver负责户代码进行DAG切分划分Stage然Stage应task调度提交Executor进行计算样Executor行执行Stagetask
（里DriverExecutor进程般分布机器）
里理解Stagetask图Spark作业划分层次：

Application户submit提交整体代码代码中action操作action算子Application划分jobjob根宽赖划分StageStage划分许（数量分区决定分区数task计算）功相task然task提交Executor进行计算执行结果返回Driver汇总存储
体现 Driver端总规划–Executor端分计算–结果汇总回Driver 思想分布式计算思想
8Spark算子
分类
方说Spark 算子致分两类
1）Transformation 变换转换算子：种变换触发提交作业完成作业中间程处理
　　　Transformation 操作延迟计算说RDD 转换生成 RDD 转换操作马执行需等 Action 操作时候会真正触发运算
2）Action 行动算子：类算子会触发 SparkContext 提交 Job 作业
　　 Action 算子会触发 Spark 提交作业（Job）数输出 Spark系统

　　方说Spark 算子致分三类
　　1）Value数类型Transformation算子种变换触发提交作业针处理数项Value型数
　　2）KeyValue数类型Transfromation算子种变换触发提交作业针处理数项KeyValue型数
　　3）Action算子类算子会触发SparkContext提交Job作业
1）Value数类型Transformation算子　　
　　输入分区输出分区型
　　　　1map算子
　　　　2flatMap算子
　　　　3mapPartitions算子
　　　　4glom算子
　　二输入分区输出分区型　
　　　　5union算子
　　　　6cartesian算子
　　三输入分区输出分区型
　　　　7grouBy算子
　　四输出分区输入分区子集型
　　　　8filter算子
　　　　9distinct算子
　　　　10subtract算子
　　　　11sample算子
　　 12takeSample算子
　　五Cache型
　　　　13cache算子　　
　　　　14persist算子
2）KeyValue数类型Transfromation算子
　　输入分区输出分区
　　　　15mapValues算子
　　二单RDD两RDD聚集
　　　单RDD聚集
　　　　16combineByKey算子
　　　　17reduceByKey算子
　　　　18partitionBy算子
　　两RDD聚集
　　　　19Cogroup算子
　　三连接
　　　　20join算子
　　　　21leftOutJoin rightOutJoin算子
　　Spark算子作详细见httpwwwcnblogscomzlslchp5723979html
3）Action算子
　　输出
　　　　22foreach算子
　　二HDFS
　　　　23saveAsTextFile算子
　　　　24saveAsObjectFile算子
　　三Scala集合数类型
　　　　25collect算子
　　　　26collectAsMap算子
　　　 27reduceByKeyLocally算子
　　　 28lookup算子
　　　　29count算子
　　　　30top算子
　　　　31reduce算子
　　　　32fold算子
　　　　33aggregate算子
　　Spark算子作详细见httpwwwcnblogscomzlslchp5723979html

1 Transformations 算子
　（1） map
　　原 RDD 数项通 map 中户定义函数 f 映射转变新元素源码中 map 算子相初始化 RDD 新 RDD 做 MappedRDD(this scclean(f))
图 1中方框表示 RDD 分区左侧分区户定义函数 fT>U 映射右侧新 RDD 分区实际等 Action算子触发 f 函数会函数stage 中数进行运算图 1 中第分区数记录 V1 输入 f通 f 转换输出转换分区中数记录 V＇1


　　　　　　图1 map 算子 RDD 转换　　　　　　　　　　　　　　　　　　　

（2） flatMap
原 RDD 中元素通函数 f 转换新元素生成 RDD 集合中元素合集合部创建 FlatMappedRDD(thisscclean(f))
　　图 2 表示 RDD 分区进行 flatMap函数操作 flatMap 中传入函数 fT>U T U 意数类型分区中数通户定义函数 f 转换新数外部方框认 RDD 分区方框代表集合 V1 V2 V3 集合作 RDD 数项存储数组容器转换V＇1 V＇2 V＇3 原数组容器结合拆散拆散数形成 RDD 中数项

图2 　flapMap 算子 RDD 转换
  （3） mapPartitions
mapPartitions 函数获取分区迭代器函数中通分区整体迭代器整分区元素进行操作部实现生成
MapPartitionsRDD图 3 中方框代表 RDD 分区图 3 中户通函数 f (iter)>iterf ilter(_>3) 分区中数进行滤等 3 数保留方块代表 RDD 分区含 1 2 3 分区滤剩元素 3

　　　　图3 mapPartitions 算子 RDD 转换
　　（4）glom
　　glom函数分区形成数组部实现返回GlommedRDD 图4中方框代表RDD分区图4中方框代表分区该图表示含V1 V2 V3分区通函数glom形成数组Array[（V1）（V2）（V3）]

　　　　　　图 4   glom算子RDD转换
（5） union
union 函数时需保证两 RDD 元素数类型相返回 RDD 数类型合 RDD 元素数类型相进行重操作保存元素果想重
distinct()时 Spark 提供更简洁 union API通 ++ 符号相 union 函数操作
图 5 中左侧方框代表两 RDD方框方框代表 RDD 分区右侧方框代表合 RDD方框方框代表分区
　　含V1V2U1U2U3U4RDD含V1V8U5U6U7U8RDD合元素形成RDDV1V1V2V8形成分区U1U2U3U4U5U6U7U8形成分区

　　图 5 union 算子 RDD 转换　

（6） cartesian
两 RDD 元素进行笛卡尔积操作操作部实现返回CartesianRDD图6中左侧方框代表两 RDD方框方框代表 RDD 分区右侧方框代表合 RDD方框方框代表分区图6中方框代表RDD方框中方框代表RDD分区
例： V1 RDD 中 W1 W2 Q5 进行笛卡尔积运算形成 (V1W1)(V1W2) (V1Q5)

图 6 cartesian 算子 RDD 转换
（7） groupBy
　　groupBy ：元素通函数生成相应 Key数转化 KeyValue 格式 Key 相元素分组
　　函数实现：
　　1）户函数预处理：
　　val cleanF scclean(f)
　　2）数 map 进行函数操作进行 groupByKey 分组操作
thismap(t > (cleanF(t) t))groupByKey(p)
　　中 p 确定分区数分区函数决定行化程度
　　图7 中方框代表 RDD 分区相key 元素合组例 V1 V2 合 V Value V1V2形成 VSeq(V1V2)

　　图 7 groupBy 算子 RDD 转换
（8） filter
filter 函数功元素进行滤元素应 f 函数返回值 true 元素 RDD 中保留返回值 false 元素滤掉部实现相生成 FilteredRDD(thisscclean(f))
面代码函数质实现：
  deffilter(fT>Boolean)RDD[T]newFilteredRDD(thisscclean(f))
　　图 8 中方框代表 RDD 分区 T 意类型通户定义滤函数 f数项操作满足条件返回结果 true 数项保留例滤掉 V2 V3 保留 V1区分命名 V＇1

　　图 8 filter 算子 RDD 转换

　　（9）distinct
　　distinctRDD中元素进行重操作图9中方框代表RDD分区通distinct函数数重例重复数V1 V1重保留份V1

　　　　图9 distinct算子RDD转换
（10）subtract
　　subtract相进行集合差操作RDD 1RDD 1RDD 2交集中元素图10中左侧方框代表两RDD方框方框代表RDD分区右侧方框
代表合RDD方框方框代表分区 V1两RDD中均根差集运算规新RDD保留V2第RDD第二RDD没新RDD元素中包含V2
　　
　　　　　　　　　　图10 subtract算子RDD转换
（11） sample
sample RDD 集合元素进行采样获取元素子集户设定否放回抽样百分机种子进决定采样方式部实现生成 SampledRDD(withReplacement fraction seed)
　　函数参数设置：
‰ 　　withReplacementtrue表示放回抽样
‰ 　　withReplacementfalse表示放回抽样
　　图 11中方框 RDD 分区通 sample 函数采样 50 数 V1 V2 U1 U2U3U4 采样出数 V1 U1 U2 形成新 RDD

　　　　　　　图11 sample 算子 RDD 转换
　　（12）takeSample
　　takeSample（）函数面sample函数原理相例采样设定采样数进行采样时返回结果RDD相采样数进行
Collect（）返回结果集合单机数组
　　图12中左侧方框代表分布式节点分区右侧方框代表单机返回结果数组通takeSample数采样设置采样份数返回结果V1

　　图12 　　takeSample算子RDD转换
　　（13） cache
cache RDD 元素磁盘缓存存相 persist(MEMORY_ONLY) 函数功
图13 中方框代表 RDD 分区左侧相数分区存储磁盘通 cache 算子数缓存存

　　　　　　图 13 Cache 算子 RDD 转换
　　（14） persist
persist 函数 RDD 进行缓存操作数缓存里 StorageLevel 枚举类型进行确定种类型组合（见10） DISK 代表磁盘MEMORY 代表存 SER 代表数否进行序列化存储
　　面函数定义 StorageLevel 枚举类型代表存储模式户通图 141 需进行选择
　　persist(newLevelStorageLevel)
　　图 141 中列出persist 函数进行缓存模式例MEMORY_AND_DISK_SER 代表数存储存磁盘序列化方式存储理

　　　　　　　　　　　　图 141 persist 算子 RDD 转换
　　图 142 中方框代表 RDD 分区 disk 代表存储磁盘 mem 代表存储存数初全部存储磁盘通 persist(MEMORY_AND_DISK) 数缓存存
分区法容纳存含 V1 V2 V3 RDD存储磁盘含U1U2RDD旧存储存

图 142 Persist 算子 RDD 转换
（15） mapValues
mapValues ：针（Key Value）型数中 Value 进行 Map 操作 Key 进行处理
图 15 中方框代表 RDD 分区 a>a+2 代表 (V11) 样 Key Value 数数 Value 中 1 进行加 2 操作返回结果 3

　　　　　　图 15 mapValues 算子 RDD 转换
（16） combineByKey
　　面代码 combineByKey 函数定义：
　　combineByKey[C](createCombiner(V) C
　　mergeValue(C V) C
　　mergeCombiners(C C) C
　　partitionerPartitioner
　　mapSideCombineBooleantrue
　　serializerSerializernull)RDD[(KC)]
说明：
‰ 　　createCombiner： V > C C 存情况通 V 创建 seq C
‰　　 mergeValue： (C V) > C C 已存情况需 merge item V
加 seq C 中者叠加
　　 mergeCombiners： (C C) > C合两 C
‰ 　　partitioner： Partitioner Shuff le 时需 Partitioner
‰ 　　mapSideCombine ： Boolean true减传输量 combine map
端先做叠加先 partition 中相 key value 叠加
shuff le
‰ 　　serializerClass： String null传输需序列化户定义序列化类：
　　例相元素 (Int Int) RDD 转变 (Int Seq[Int]) 类型元素 RDD图 16中方框代表 RDD 分区图通 combineByKey (V12) (V11)数合（ V1Seq(21)）
　　
　　　　　　图 16 comBineByKey 算子 RDD 转换
（17） reduceByKey
reduceByKey combineByKey 更简单种情况两值合成值（ Int Int V）to （Int Int C）叠加 createCombiner reduceBykey 简单直接返回 v mergeValue mergeCombiners 逻辑相没区
函数实现：
def reduceByKey(partitioner Partitioner func (V V) > V) RDD[(K V)]
{
combineByKey[V]((v V) > v func func partitioner)
}
　　图17中方框代表 RDD 分区通户定义函数 (AB) > (A + B) 函数相 key 数 (V12) (V11) value 相加运算结果（ V13）

　　　　　　　　图 17 reduceByKey 算子 RDD 转换
（18）partitionBy
　　partitionBy函数RDD进行分区操作
　　函数定义
　　partitionBy（partitioner：Partitioner）
　　果原RDD分区器现分区器（partitioner）致重分区果致相根分区器生成新ShuffledRDD
　　图18中方框代表RDD分区通新分区策略原分区V1 V2数合分区

　　　　图18　　partitionBy算子RDD转换
（19）Cogroup
　　cogroup函数两RDD进行协划分cogroup函数定义
　　cogroup[W]（other： RDD[（K W）] numPartitions： Int）： RDD[（K （Iterable[V] Iterable[W]））]
　　两RDD中KeyValue类型元素RDD相Key元素分聚合集合返回两RDD中应Key元素集合迭代器
　　（K （Iterable[V] Iterable[W]））
　　中KeyValueValue两RDD相Key两数集合迭代器构成元组
　　图19中方框代表RDD方框方框代表RDD中分区 RDD1中数（U11）（U12）RDD2中数（U12）合（U1（（12）（2）））

　　　　　　　　图19 Cogroup算子RDD转换
（20） join
join 两需连接 RDD 进行 cogroup函数操作相 key 数够放分区 cogroup 操作形成新 RDD key 元素进行笛卡尔积操作返回结果展应 key 元组形成集合返回 RDD[(K (V W))]
　　面代码 join 函数实现质通 cogroup 算子先进行协划分通 flatMapValues 合数散
thiscogroup(otherpartitioner)f latMapValues{case(vsws) > for(v图 20两 RDD join 操作示意图方框代表 RDD方框代表 RDD 中分区函数相 key 元素 V1 key 做连接结果 (V1(11)) (V1(12))

　　　　　　　　　　　　　　　　　　　　图 20 join 算子 RDD 转换
（21）eftOutJoinrightOutJoin
　　LeftOutJoin（左外连接）RightOutJoin（右外连接）相join基础先判断侧RDD元素否空果空填充空果空数进行连接运算
返回结果
面代码leftOutJoin实现
if （wsisEmpty） {
vsmap（v > （v None））
} else {
for （v < vs w < ws） yield （v Some（w））
}

2 Actions 算子
　　质 Action 算子中通 SparkContext 进行提交作业 runJob 操作触发RDD DAG 执行
例 Action 算子 collect 函数代码感兴趣读者着入口进行源码剖析：

**
* Return an array that contains all of the elements in this RDD
*
def collect() Array[T] {
* 提交 Job*
val results scrunJob(this (iter Iterator[T]) > itertoArray)
Arrayconcat(results _*)
}
（22） foreach
　　foreach RDD 中元素应 f 函数操作返回 RDD Array 返回Uint图22表示 foreach 算子通户定义函数数项进行操作例中定义函数 println()控制台印数项
　　
　　　　　　图 22 foreach 算子 RDD 转换
　　（23） saveAsTextFile
　　函数数输出存储 HDFS 指定目录
面 saveAsTextFile 函数部实现部
　　通调 saveAsHadoopFile 进行实现：
thismap(x > (NullWritableget() new Text(xtoString)))saveAsHadoopFile[TextOutputFormat[NullWritable Text]](path)
RDD 中元素映射转变 (null xtoString)然写入 HDFS
　　图 23中左侧方框代表 RDD 分区右侧方框代表 HDFS Block通函数RDD 分区存储 HDFS 中 Block
　　
　　　　　　　　　　　　图 23 saveAsHadoopFile 算子 RDD 转换
（24）saveAsObjectFile
　　saveAsObjectFile分区中10元素组成Array然Array序列化映射（NullBytesWritable（Y））元素写入HDFSSequenceFile格式
　　面代码函数部实现
　　map（x>（NullWritableget（）new BytesWritable（Utilsserialize（x））））
　　图24中左侧方框代表RDD分区右侧方框代表HDFSBlock 通函数RDD分区存储HDFSBlock

　　　　　　　　　　　　图24 saveAsObjectFile算子RDD转换

（25） collect
　　collect 相 toArray toArray 已时推荐 collect 分布式 RDD 返回单机 scala Array 数组数组运 scala 函数式操作
　　图 25中左侧方框代表 RDD 分区右侧方框代表单机存中数组通函数操作结果返回 Driver 程序节点数组形式存储

　　图 25 Collect 算子 RDD 转换　　
（26）collectAsMap
　　collectAsMap（KV）型RDD数返回单机HashMap 重复KRDD元素面元素覆盖前面元素
　　图26中左侧方框代表RDD分区右侧方框代表单机数组数通collectAsMap函数返回Driver程序计算结果结果HashMap形式存储

　　　　　　　　　　图26 CollectAsMap算子RDD转换

（27）reduceByKeyLocally
　　实现先reducecollectAsMap功先RDD整体进行reduce操作然收集结果返回HashMap
（28）lookup
面代码lookup声明
lookup（key：K）：Seq[V]
Lookup函数（KeyValue）型RDD操作返回指定Key应元素形成Seq 函数处理优化部分果RDD包含分区器会应处理K分区然返回（KV）形成Seq 果RDD包含分区器需全RDD元素进行暴力扫描处理搜索指定K应元素
　　图28中左侧方框代表RDD分区右侧方框代表Seq结果返回Driver节点应中

　　　　　　图28 lookupRDD转换
（29） count
　　count 返回整 RDD 元素数
　　部函数实现：
　　defcount()LongscrunJob(thisUtilsgetIteratorSize_)sum
　　图 29中返回数数 5方块代表 RDD 分区

　　　　图29 count RDD 算子转换
（30）top
top返回k元素函数定义
top（num：Int）（implicit ord：Ordering[T]）：Array[T]
相函数说明
·top返回k元素
·take返回k元素
·takeOrdered返回k元素返回数组中保持元素序
·first相top（1）返回整RDD中前k元素定义排序方式Ordering[T]
返回含前k元素数组

（31）reduce
　　reduce函数相RDD中元素进行reduceLeft函数操作函数实现
　　Some（iterreduceLeft（cleanF））
　　reduceLeft先两元素进行reduce函数操作然结果迭代器取出元素进行reduce函数操作直迭代器遍历完元素结果RDD中先分区中元素集合分进行reduceLeft 分区形成结果相元素结果集合进行reduceleft操作
　　例：户定义函数
　　f：（AB）>（A_1+@+B_1A_2+B_2）
　　图31中方框代表RDD分区通户定函数f数进行reduce运算示例
返回结果V1@[1]V2U@U2@U3@U412

图31 reduce算子RDD转换
（32）fold
　　foldreduce原理相reduce相reduce时迭代器取第元素zeroValue
　　图32中通面户定义函数进行fold运算图中方框代表RDD分区读者参reduce函数理解
　　fold（（V0@2））（（AB）>（A_1+@+B_1A_2+B_2））

　　　　　　　　　　图32 fold算子RDD转换
（33）aggregate
　　aggregate先分区元素进行aggregate操作分区结果进行fold操作
　　aggreagatefoldreduce处aggregate相采方式进行数聚集种聚集行化 foldreduce函数运算程中分区中需进行串行处理分区串行计算完结果结果前方式进行聚集返回终聚集结果
　　函数定义
aggregate[B]（z： B）（seqop：（BA） > Bcombop：（BB） > B）： B
　　图33通户定义函数RDD 进行aggregate聚集操作图中方框代表RDD分区
　　rddaggregate（V0@2）（（AB）>（A_1+@+B_1A_2+B_2））（AB）>（A_1+@+B_1A_@+B_2））
　　介绍两计算模型中两特殊变量
　　广播（broadcast）变量：广泛广播Map Side Join中表广播变量等场景数集合单节点存够容纳需RDD样节点间散存储
Spark运行时广播变量数发节点保存续计算复相Hadoodistributed cache广播容跨作业享 Broadcast底层实现采BT机制

　　　　　　　　图33 aggregate算子RDD转换
②代表V
③代表U
accumulator变量：允许做全局累加操作accumulator变量广泛应中记录
前运行指标情景

详细介绍
官方文档列举32种常见算子包括Transformation20种操作Action12种操作
（注：截图windows运行结果）
Transformation：
1map
map输入变换函数应RDD中元素mapPartitions应分区区mapPartitions调粒度parallelize（1 to 10 3）map函数执行10次mapPartitions函数执行3次

2filter（function）
滤操作满足filterfunction函数trueRDD元素组成新数集：filter（a 1）

3flatMap（function）
mapRDD中元素逐进行函数操作映射外RDDflatMap操作函数应RDD中元素返回迭代器容构成新RDDflatMap操作函数应RDD中元素返回迭代器容构成RDD
flatMapmap区map映射flatMap先映射扁化map次（func）产生元素返回象flatMap步象合象

4mapPartitions（function）
区foreachPartition（属Action返回值）mapPartitions获取返回值map区前面已提单独运行RDD分区（block）类型TRDD运行时（function）必须Iterator > Iterator类型方法（入参）

5mapPartitionsWithIndex（function）
mapPartitions类似需提供表示分区索引值整型值作参数function必须（int Iterator）>Iterator类型

6sample（withReplacement fraction seed）
采样操作样中取出部分数withReplacement否放回fraction采样例seed指定机数生成器种子（否放回抽样分truefalsefraction取样例(0 1]seed种子整型实数）

7union（otherDataSet）
源数集数集求集重

8intersection（otherDataSet）
源数集数集求交集重序返回

9distinct（[numTasks]）
返回源数集重新数集重局部序整体序返回（详细介绍见
httpsblogcsdnnetFortuna_iarticledetails81506936）
注：groupByKeyreduceByKeyaggregateByKeysortByKeyjoincogroup等Transformation操作均包含[numTasks]务数参数参考行链接理解

注：pairRDD进行操作添加pairRDD简易创建程

10groupByKey([numTasks])
PairRDD（kv）RDD调返回（kIterable）作相键值分组集合序列中序确定groupByKey键值集合加载存中存储计算键应值太易导致存溢出
前求集union方法pair1pair2变相键值pair3进行groupByKey

11reduceByKey（function[numTasks]）
groupByKey类似(a1) (a2) (b1) (b2)groupByKey产生中间结果( (a1) (a2) ) ( (b1) (b2) )reduceByKey(a3) (b3)
reduceByKey作聚合groupByKey作分组（functionkey值进行聚合）

12aggregateByKey（zeroValue）（seqOp combOp [numTasks]）
类似reduceByKeypairRDD中想key值进行聚合操作初始值（seqOp中combOpenCL中未）应返回值pairRDD区aggregate（返回值非RDD）

13sortByKey（[ascending] [numTasks]）
样基pairRDD根key值进行排序ascending升序默认true升序numTasks

14join（otherDataSet[numTasks]）
加入RDD（kv）（kw）类型dataSet调返回（k（vw））pair dataSet

15cogroup（otherDataSet[numTasks]）
合两RDD生成新RDD实例中包含两Iterable值第表示RDD1中相值第二表示RDD2中相值（key值）操作需通partitioner进行重新分区需执行次shuffle操作（两RDD前进行shuffle需）

16cartesian（otherDataSet）
求笛卡尔积该操作会执行shuffle操作

17pipe（command[envVars]）
通shell命令RDD分区进行道化通pipe变换shell命令Spark中生成新RDD：

（图莫怪^_^）
18coalesce（numPartitions）
重新分区减少RDD中分区数量numPartitions

19repartition（numPartitions）
repartitioncoalesce接口中shuffletrue简易实现Reshuffle RDD机分区分区数量衡分区分区数远原分区数需shuffle

20repartitionAndSortWithinPartitions（partitioner）
该方法根partitionerRDD进行分区结果分区中key进行排序

Action：
1reduce（function）
reduceRDD中元素两两传递输入函数时产生新值新值RDD中元素传递输入函数直值止

2collect（）
RDDArray数组形式返回中元素（具体容参见：
httpsblogcsdnnetFortuna_iarticledetails80851775）

3count（）
返回数集中元素数默认Long类型

4first（）
返回数集第元素（类似take(1)）

5takeSample（withReplacement num [seed]）
数集进行机抽样返回包含num机抽样元素数组withReplacement表示否放回抽样参数seed指定生成机数种子
该方法仅预期结果数组情况数加载driver端存中

6take（n）
返回包含数集前n元素数组（0标n1标元素）排序

7takeOrdered（n[ordering]）
返回RDD中前n元素默认序排序（升序）者定义较器序排序

8saveAsTextFile（path）
dataSet中元素文文件形式写入文件系统者HDFS等Spark元素调toString方法数元素转换文文件中行记录
文件保存文件系统会保存executor机器目录

9saveAsSequenceFile（path）（Java and Scala）
dataSet中元素Hadoop SequenceFile形式写入文件系统者HDFS等（pairRDD操作）

10saveAsObjectFile（path）（Java and Scala）
数集中元素ObjectFile形式写入文件系统者HDFS等

11countByKey（）
统计RDD[KV]中K数量返回具key计数（kint）pairshashMap

12foreach（function）
数集中元素运行函数function

补充：Spark23官方文档中原[numTasks]务数参数改[numPartitions]分区数
实践
SparkRDD算子分两类：TransformationAction
Transformation：延迟加载数Transformation会记录元数信息计算务触发Action时会真正开始计算
Action：立加载数开始计算
创建RDD方式两种：
1通sctextFile(rootwordstxt)文件系统中创建 RDD
2#通行化scala集合创建RDD：val rdd1 scparallelize(Array(12345678))
1简单算子说明
里先说简单Transformation算子
通行化scala集合创建RDD
val rdd1 scparallelize(Array(12345678))
查该rdd分区数量
rdd1partitionslength
map方法scala中样List中数出做函数运算
sortBy：数进行排序
val rdd2 scparallelize(List(56473829110))map(_*2)sortBy(x>xtrue)
filter：List中数进行函数造作挑选出10值
val rdd3 rdd2filter(_>10)
collect：终结果显示出
flatMap数先进行map操作进行flat（碾压）操作
rdd4flatMap(_split(’ ))collect
运行效果图

val rdd1 scparallelize(List(56473829110))
val rdd2 scparallelize(List(56473829110))map(_*2)sortBy(x>xtrue)
val rdd3 rdd2filter(_>10)
val rdd2 scparallelize(List(56473829110))map(_*2)sortBy(x>x+true)
val rdd2 scparallelize(List(56473829110))map(_*2)sortBy(x>xtoStringtrue)

intersection求交集
val rdd9 rdd6intersection(rdd7)
val rdd1 scparallelize(List((tom 1) (jerry 2) (kitty 3)))
val rdd2 scparallelize(List((jerry 9) (tom 8) (shuke 7)))

join
val rdd3 rdd1join(rdd2)

val rdd3 rdd1leftOuterJoin(rdd2)

val rdd3 rdd1rightOuterJoin(rdd2)

union：求集注意类型致
val rdd6 scparallelize(List(5647))
val rdd7 scparallelize(List(1234))
val rdd8 rdd6union(rdd7)
rdd8distinctsortBy(x>x)collect

groupByKey
val rdd3 rdd1 union rdd2
rdd3groupByKey
rdd3groupByKeymap(x>(x_1x_2sum))

cogroup
val rdd1 scparallelize(List((tom 1) (tom 2) (jerry 3) (kitty 2)))
val rdd2 scparallelize(List((jerry 2) (tom 1) (shuke 2)))
val rdd3 rdd1cogroup(rdd2)
val rdd4 rdd3map(t>(t_1 t_2_1sum + t_2_2sum))

cartesian笛卡尔积
val rdd1 scparallelize(List(tom jerry))
val rdd2 scparallelize(List(tom kitty shuke))
val rdd3 rdd1cartesian(rdd2)

接说简单Action算子
val rdd1 scparallelize(List(12345) 2)
#collect
rdd1collect
#reduce
val rdd2 rdd1reduce(+)
#count
rdd1count
#top
rdd1top(2)
#take
rdd1take(2)
#first(similer to take(1))
rdd1first
#takeOrdered
rdd1takeOrdered(3)

2复杂算子说明
mapPartitionsWithIndex partition中分区号应值出源码
val func (index Int iter Iterator[(Int)]) > {
itertoListmap(x > [partID + index + val + x + ])iterator
}
val rdd1 scparallelize(List(123456789) 2)
rdd1mapPartitionsWithIndex(func)collect

aggregate
def func1(index Int iter Iterator[(Int)]) Iterator[String] {
itertoListmap(x > [partID + index + val + x + ])iterator
}
val rdd1 scparallelize(List(123456789) 2)
rdd1mapPartitionsWithIndex(func1)collect
###action操作第参数初始值二2函数(第函数先分区进行合第二函数分区合结果进行合)
###0 + (0+1+2+3+4 + 0+5+6+7+8+9)
rdd1aggregate(0)(_+_ _+_)
· 1

rdd1aggregate(0)(mathmax( ) _ + _)
###0分01分区List元素分区中值里分37然0+3+710

###51 52345 –> 567899 –> 5 + (5+9)
rdd1aggregate(5)(mathmax( ) _ + _)

val rdd3 scparallelize(List(12233454567)2)
rdd3aggregate()((xy) > mathmax(xlength ylength)toString (xy) > x + y)
######### length分两分区元素length进行较0分区字符串21分区字符串4然结果返回分先结果2442

val rdd4 scparallelize(List(1223345)2)
rdd4aggregate()((xy) > mathmin(xlength ylength)toString (xy) > x + y)
######## length012较字符串0然字符串023较值1

aggregateByKey
val pairRDD scparallelize(List( (cat2) (cat 5) (mouse 4)(cat 12) (dog 12) (mouse 2)) 2)
def func2(index Int iter Iterator[(String Int)]) Iterator[String] {
itertoListmap(x > [partID + index + val + x + ])iterator
}
pairRDDmapPartitionsWithIndex(func2)collect

pairRDDaggregateByKey(0)(mathmax( ) _ + _)collect
########## 先0号分区中数进行操作（初始值数进行较）（cat5）(mouse4)然1号分区中数进行操作（cat12）（dog12）(mouse2)然两分区数进行相加终结果

coalesce
#coalesce(2 false)代表数重新分成2区进行shuffle（数重新进行机分配数通网络分配机器）
val rdd1 scparallelize(1 to 10 10)
val rdd2 rdd1coalesce(2 false)
rdd2partitionslength

repartition
repartition效果等coalesce(x true)

collectAsMap Map(b > 2 a > 1)
val rdd scparallelize(List((a 1) (b 2)))
rddcollectAsMap

combineByKey reduceByKey相效果
###第参数x原封动取出第二参数函数局部运算第三函数局部运算结果做运算
###分区中key中value中第值 (hello1)(hello1)(good1)–>(hello(11)good(1))–>x相hello第1 good中1
val rdd1 sctextFile(hdfsmaster9000wordcountinput)flatMap(split( ))map(( 1))
val rdd2 rdd1combineByKey(x > x (a Int b Int) > a + b (m Int n Int) > m + n)
rdd1collect

###input3文件时(3block块分三区 3文件3block ) 会加310
val rdd3 rdd1combineByKey(x > x + 10 (a Int b Int) > a + b (m Int n Int) > m + n)
rdd3collect

val rdd4 scparallelize(List(dogcatgnusalmonrabbitturkeywolfbearbee) 3)
val rdd5 scparallelize(List(112221222) 3)
val rdd6 rdd5zip(rdd4)

第参数List(_)代表第元素转换List第二参数x List[String] y String) > x + y代表元素y加入list中第三参数(m List[String] n List[String]) > m ++ n)代表两分区list合成新List
val rdd7 rdd6combineByKey(List(_) (x List[String] y String) > x + y (m List[String] n List[String]) > m ++ n)

countByKey
val rdd1 scparallelize(List((a 1) (b 2) (b 2) (c 2) (c 1)))
rdd1countByKey
rdd1countByValue

filterByRange
val rdd1 scparallelize(List((e 5) (c 3) (d 4) (c 2) (a 1)))
val rdd2 rdd1filterByRange(b d)
rdd2collect

flatMapValues Array((a1) (a2) (b3) (b4))
val rdd3 scparallelize(List((a 1 2) (b 3 4)))
val rdd4 rdd3flatMapValues(_split( ))
rdd4collect

foldByKey
val rdd1 scparallelize(List(dog wolf cat bear) 2)
val rdd2 rdd1map(x > (xlength x))
val rdd3 rdd2foldByKey()(+)

keyBy 传入参数做key
val rdd1 scparallelize(List(dog salmon salmon rat elephant) 3)
val rdd2 rdd1keyBy(_length)
rdd2collect

keys values
val rdd1 scparallelize(List(dog tiger lion cat panther eagle) 2)
val rdd2 rdd1map(x > (xlength x))
rdd2keyscollect
rdd2valuescollect

方法英文解释
#
map(func)
Return a new distributed dataset formed by passing each element of the source through a function func
filter(func)
Return a new dataset formed by selecting those elements of the source on which func returns true
flatMap(func)（部执行序右左先执行Map执行Flat）
Similar to map but each input item can be mapped to 0 or more output items (so func should return a Seq rather than a single item)
mapPartitions(func)
Similar to map but runs separately on each partition (block) of the RDD so func must be of type Iterator > Iterator when running on an RDD of type T
mapPartitionsWithIndex(func)
Similar to mapPartitions but also provides func with an integer value representing the index of the partition so func must be of type (Int Iterator) > Iterator when running on an RDD of type T
sample(withReplacement fraction seed)
Sample a fraction fraction of the data with or without replacement using a given random number generator seed
union(otherDataset)
Return a new dataset that contains the union of the elements in the source dataset and the argument
intersection(otherDataset)
Return a new RDD that contains the intersection of elements in the source dataset and the argument
distinct([numTasks]))
Return a new dataset that contains the distinct elements of the source dataset
groupByKey([numTasks])
When called on a dataset of (K V) pairs returns a dataset of (K Iterable) pairs
reduceByKey(func [numTasks])
When called on a dataset of (K V) pairs returns a dataset of (K V) pairs where the values for each key are aggregated using the given reduce function func which must be of type (VV) > V Like in groupByKey the number of reduce tasks is configurable through an optional second argument
aggregateByKey(zeroValue)(seqOp combOp [numTasks])
When called on a dataset of (K V) pairs returns a dataset of (K U) pairs where the values for each key are aggregated using the given combine functions and a neutral zero value Allows an aggregated value type that is different than the input value type while avoiding unnecessary allocations Like in groupByKey the number of reduce tasks is configurable through an optional second argument
sortByKey([ascending] [numTasks])
When called on a dataset of (K V) pairs where K implements Ordered returns a dataset of (K V) pairs sorted by keys in ascending or descending order as specified in the boolean ascending argument
join(otherDataset [numTasks])
When called on datasets of type (K V) and (K W) returns a dataset of (K (V W)) pairs with all pairs of elements for each key Outer joins are supported through leftOuterJoin rightOuterJoin and fullOuterJoin
cogroup(otherDataset [numTasks])
When called on datasets of type (K V) and (K W) returns a dataset of (K (Iterable Iterable)) tuples This operation is also called groupWith
cartesian(otherDataset)
When called on datasets of types T and U returns a dataset of (T U) pairs (all pairs of elements)
pipe(command [envVars])
Pipe each partition of the RDD through a shell command eg a Perl or bash script RDD elements are written to the process’s stdin and lines output to its stdout are returned as an RDD of strings
coalesce(numPartitions)
Decrease the number of partitions in the RDD to numPartitions Useful for running operations more efficiently after filtering down a large dataset
repartition(numPartitions)
Reshuffle the data in the RDD randomly to create either more or fewer partitions and balance it across them This always shuffles all data over the network
repartitionAndSortWithinPartitions(partitioner)
Repartition the RDD according to the given partitioner and within each resulting partition sort records by their keys This is more efficient than calling repartition and then sorting within each partition because it can push the sorting down into the shuffle machinery
(K(IterableIterable))

9Spark RDD
着时间推移数分析已达新程度反改变运作模式期天数分析仅处理量数具快速周转时间定目标然Hadoop数分析背伦技术快速处理方面存足着Spark出现数处理速度便更期
谈Spark时想第术语弹性分布式数集(RDD)Spark RDD数处理更快外Spark关键特性支持计算期间数集进行逻辑分区
文中讨Spark RDD技术方面进步解Spark RDD底层技术细节外概述RDDSpark中
Spark RDD特性
RDD定义Resilient Distributed Dataset(弹性分布式数集)中术语表示特性
· Resilient 通RDD谱系图(DAG)实现容错节点发生障时进行重新计算
· Distributed  Spark RDD数集驻留节点中
· Dataset 您数记录
Hadoop设计中RDD挑战然Spark RDD解决方案似非常高效取决惰性计算Spark中RDDs需工作节省量数处理时间整程效率
Hadoop Mapreduce通特性克服Spark RDD许缺点Spark RDD流行原
Spark RDD核心特性
· 存计算
· 惰性计算
· 容错
· 变性
· 分区
· 持久性
· 粗粒度操作
· 位置粘性
节中逐步讨问题

Spark RDD种表示分布节点数集技术行操作换句话说Spark RDDApache Spark容错抽象Apache Spark基数结构
Spark中RDD变分布式象集合支持两种方法
· cache()
· persist()

Spark RDD存缓存技术Spark RDD中数集进行逻辑分区存缓存处果数合适会余数发送磁盘进行重新计算什称弹性您需时Spark中提取RDD整数处理更快
Spark数处理方面Hadoop快100倍面Apache Spark更快素

Spark RDD支持操作
Spark中RDD支持两种类型操作
1 Transformations
2 Actions

Transformation
transformation情况Spark RDD现数集创建新数集引Spark RDD转换示例说map转换通函数传递dataset元素作返回值发送表示结果新RDD
Scala
val l sctextFile(exampletxt)

val lLengths lmap(s > slength)

val totalLength lLengthsreduce((a b) > a + b)
果想llengthpersist()函数示
lLengths persist()
您httpssparkapacheorg参考API文档获Spark RDD支持转换详细列表

Spark RDD支持两种类型转换
1 Narrow transformation
2 Wide transformation
Narrow transformation情况输出RDD父RDD数分区相关联广泛转换中输出RDD许父RDD分区结果换句话说谓shuffle transformation

Spark RDD转换惰性会立计算结果相反记住基数集应转换数集引文件示例示操作需结果时转换Spark RDD中计算进导致更快更效数处理

次转换RDD运行操作时会Spark RDD中进行重新计算persist方法Spark元素保存集群中便次查询时更快访问支持磁盘持久存储Spark RDDs跨节点进行复制

Actions
操作期间RDD数集执行计算值返回驱动程序例reduce某函数聚合RDD元素终结果返回程序操作
创建Spark RDD三程
1 行集合
2 外部数集(外部存储系统享文件系统HBaseHDFS)
3 现Apache Spark RDDs
接讨方法中种解创建Spark RDDs

弹性分布式数集(RDD)Apache Spark重特性非常重解Apache Spark数行业中重性

行集合
您通JavaScalaPython中现驱动程序集合调SparkContext接口parallelize方法创建行集合例中复制集合元素构成分布式数集行操作

Scala中行化集合Spark RDD例子
数字26保存行集合：
val collection Array(2 3 4 56)

val prData sparksparkContextparallelize(collection)

里创建分布式数集prData够行操作您调prDatareduce()数组中元素相加

行化集合关键参数决定数集分割成分区号例中Spark集群分区运行单务通常集群中单CPU 24分区理想Spark会根集群动设置分区数量户通作行化第二参数传递手动设置

外部数库
Apache SparkHadoop支持文件存储创建分布式数集中包括
· Local file system
· HDFS
· Cassandra
· HBase
· Amazon S3
Spark支持类似文件格式：
· Text files
· Sequence Files
· CSV
· JSON
· Any Hadoop Input Format
例SparkContext接口textFile方法创建文文件Spark RDDs方法接受文件URL系统路径hdfs等等)文件作行集合读取

里重素果文件系统路径必须节点相路径访问该文件必须数文件复制节点需网络挂载享文件系统

您数帧读取器接口加载外部数集然 RDD方法数集转换RDD

面文文件转换示例稍返回字符串数集
val exDataRDD sparkreadtextFile(pathoftextfile)rdd

现RDDS
RDDS变改变transformation您现RDD创建新RDD没突变发生变化集群中保持致性目操作少
· map
· filter
· count
· distinct
· flatmap
例：
val seasons sparksparkContextparallelize(Seq(summer monsoon spring winter))

val seasons1 seasonsmap(s > (scharAt(0) s))

关系型数库数性优化解决方案分表(前表历史表)表分区数清理原
原目
· 交易量者日积月累造成数库数量越越会导致系统性幅降部分业务表数作备份清理
· 减少数量提升请求响应速度提升户体验
数否需清理阀值判断
通常表磁盘超 5GB OLTP 系统(联机事务处理)表记录超 3000 万应考虑表进行分区者分表
述阀值外根数库性指标情况考虑分区者分表已充分挖掘表设计索引设计查询设计等单表性优化手段然满足业务需时候表容量记录尚未达述阀值考虑分区者分表时间点记录数该表阀值
般讲记录数标记阀值会表磁盘容量更容易操作般达该阀值时记录数作阀值文会作阀值表数格式索引设计单位事务处理时间容忍度阀值
满负载周期判断
说张空表需久达阀值周期称满负载周期
果业务量已稳定数量积累前阀值需时间该表满负载周期果业务量稳定升数量递增根递增速度估算出满负载周期——说什时候达阀值什时候进行迁移做提前规划心中数
迁移周期判断
该表阀值需迁移周期进行判断久该表迁移次？
天迁移历史表次迁移周期天样操作简单完全做成定时务系统干频繁
三分法满负载周期找出 13 周期数进行迁移迁移周期 13 满负载周期样然较计划性操作稍微复杂(特业务量稳定时候)次迁移需开发运维计划参
数流历史
前表 > 历史表 > 备份表

图示绿色代表前表表示然存高频率写入操作表查询性非常高黄色代表历史表表示写入频率已查询需求查询性高灰色代表备份表表示数张表中已提供写入操作视情况会提供查询操作查询性般
类型数分区方案
· 联机交易(实时类交易写频繁)提供日交易查询建立AB表进行日切日切换提供服务张表进行迁移迁移联机交易进入联机历史表原表(A B 表)执行 truncate 操作历史表交易创建时间(更新时间完成时间)进行 range 分区操作(月分区具体业务量递增情况)
· 清分交易(非实时类交易读频繁)供日外联机交易查询台查询清分等数操作月清分交易进行分表操作样做话前端应需清分查询进行数库路跨月查询结果进行整合滤清分表建立前表历史表机制根迁移周期进行定期迁移
· 日志类数非实时类交易
· 户认证信息(访问度高读写)根业务查询需考虑时间进行 range 分区(推荐 range扩展性较高般讲业务量暴增满足需)者户 id 进行 hash range 分区分区字段选取参考注意点中列举事项
历史表清理方案
· 交易类(包括联机清分分润)数5年清理次
· 日志类数5年清理次
· 户认证商户认证类数直保留
注意点
· 日志类表(操作日志系统日志结果记录务记录等)分区处理：时间进行分区
· 交易类表(流水明细账差异等)分区处理：创建时间更新时间进行分区
· 通知类表分区处理：时间进行分区
· 分区字段选取：般常查询字段进行分区样会助提高查询速度建议 id 进行分区非业务系统里固定 id 查询特否仅分区索引浪费会没分区慢
· range 分区 hash 分区做分区时候考虑分区扩展性原分区 2 年应该考虑重新分区事情分区期根业务量增长情况加 2 年分区…类推
· 分区数较均匀太太少根分区字段快定位分区范围
· 具体分区数量少合适？原数范围缩做全盘扫描会慢时候佳视具体情况十万百万等
· 相关业务操作(SQL)量分区部完成必须跨分区提取话建议行提取提高速度

数仓库缓慢变化维(Slow changing demenison) 实现方案
缓慢变化维定义
Wikipedia中定义：
Dimension is a term in data management and data warehousing that refers to logical groupings of data such as geographical location customer information or product information
Slowly Changing Dimensions (SCD) are dimensions that have data that slowly changes
意说数会发生缓慢变化维度缓慢变化维
举例子清楚：
零售业数仓库中事实表保存着销售员销售记录某天销售员北京分公司调海分公司保存变化呢？说销售员维度恰处理变化先回答问题什处理保存变化？果统计北京区海区总销售情况时候销售员销售记录应该算北京算海？然调离前算北京调离算海标记销售员属区域？里需处理维度数缓慢变化维需做事情
处理缓慢变化维般情况种解决方案：
新数覆盖旧数

方法必须前提条件关心数剧变化例某销售员英文名改果关心员工英文名什变化直接覆盖(修改)数仓库中数

二保存条记录添加字段加区分
种情况直接新添条记录时保留原记录单独专字段保存区：
(表格中Supplier_State表示面例子中属区域描述清晰代理键表示)

Supplier_key Supplier_Code Supplier_Name Supplier_State Disable
001 ABC Phlogistical Supply Company CA Y
002 ABC Phlogistical Supply Company IL N
：

Supplier_key Supplier_Code Supplier_Name Supplier_State Version
001 ABC Phlogistical Supply Company CA 0
002 ABC Phlogistical Supply Company IL 1

两种添加数版信息否标识新旧数
面种添加记录生效日期失效日期标识新旧数：

Supplier_key Supplier_Code Supplier_Name Supplier_State Start_Date End_Date
001 ABC Phlogistical Supply Company CA 01Jan2000 21Dec2004
002 ABC Phlogistical Supply Company IL 22Dec2004

空End_Date表示前版数者默认时间 ( 12319999)代空值样数索引识

三字段保存值

Supplier_key Supplier_Name Original_Supplier_State Effective_Date Current_Supplier_State
001 Phlogistical Supply Company CA 22Dec2004 IL

种方法字段保存变化痕迹种方法象第二种方法样保存变化记录保存两次变化记录适变化超两次维度

四外建表保存历史记录

外建历史表表存变化历史记录维度保存前数

Supplier
Supplier_key Supplier_Name Supplier_State
001 Phlogistical Supply Company IL

Supplier_History
Supplier_key Supplier_Name Supplier_State Create_Date
001 Phlogistical Supply Company CA 22Dec2004

种方法仅仅记录变化历史痕迹实做起统计运算方便

五混合模式
种模式种模式混合体相言种方法更全面更应错综复杂易变化户需求较常

Row_Key Supplier_key Supplier_Code Supplier_Name Supplier_State Start_Date End_Date Current Indicator
1 001 ABC001 Phlogistical Supply Company CA 22Dec2004 15Jan2007 N
2 001 ABC001 Phlogistical Supply Company IL 15Jan2007 1Jan2099 Y

中方法条优点：
1 简单滤条件选出维度前值
2 较容易关联出历史意时刻事实数值
3 果事实表中时间字段(：Order Date Shipping Date Confirmation Date)容易选择条维度数进行关联分析

中Row_Key Current Indicator字段加更方便毕竟维度表数点冗余字段占太空间提高查询效率
种设计模式事实表应Supplier_key外键然字段唯标识条维度数形成事实表维表关系做事实维度做关联时应加时间戳字段(Indicator字段)

六非常规混合模式
面说第五种实现方式点弊端事实表维表关系关系种关系建模时解决报表层面报表运行时解决BI语意层建模时需添加时间滤条件较繁琐
面种解决方案解决关系修改事实表：

Supplier Dimension
Version_Number Supplier_key Supplier_Code Supplier_Name Supplier_State Start_Date End_Date
1 001 ABC001 Phlogistical Supply Company CA 22Dec2004 15Jan2007
0 001 ABC001 Phlogistical Supply Company IL 15Jan2007 1Jan2099

Fact Delivery (描述清晰样代理键标识维度)
Delivery_Key Supplier_key Supplier_version_number Quantity Product Delivery_Date Order_Date
1 001 0 132 Bags 22Dec2006 15Oct2006
2 001 0 324 Chairs 15Jan2007 1Jan2007

方案中维表中前数版号始终0插入维度数时先老版数version_number改成1（递增）然插入前数时保持前数版号始终0
事实表中插入数时维度数版号始终全部0
方案完全解决事实表维表关系问题外优点保证事实表维表参完整性ERwinPowerDesigner等建模工具建模时Version_NumberSupplier_key作复合键两实体间建立链接

MySQLTeradataPySpark代码互转表代码
代码描述
MySQL
Teradata SQL
PySpark
添加删列
1添加列
alter table [`<架构名称>`] `<表名>` add column <字段名> <类型>

2删列
alter table [`<架构名称>`] `<表名>` drop column <字段名>
1添加列
ALTER TABLE [<架构名称>]<表名> ADD <字段名> <类型>

2删列
ALTER TABLE [<架构名称>]<表名> DROP <字段名>
1添加列
…
withColumn(＇<字段名>＇ sum( [col] for col in columns))

2删列
drop(＇<字段名>＇)
删库
DROP DATABASE IF EXISTS] <库名>
DELETE DATABASE <库名> ALL
Parquet文件中：
import subprocess

subprocesscheck_call(＇rm r <存储路径>＇)shellTrue)

Hive表中：
from pysparksql import HiveContext
hive HiveContext(sparksparkContext)
hivesql(＇drop database if exists <库名> cascade＇)
删表
DROP TABLE [`<架构名称>`] `<表名>`
DROP TABLE [<架构名称>]<表名>
Parquet文件中：
import subprocess

subprocesscheck_call(＇rm r <存储路径><表名>＇)shellTrue)

Hive表中：
from pysparksql import HiveContext
hive HiveContext(sparksparkContext)

hivesql(＇drop table if exists [`<库名>`]`<表名>` purge＇)
清表中数
TUNCATE TABLE [`<架构名称>`] `<表名>`
DELETE [<架构名称>]<表名> ALL
Parquet文件中：
import subprocess
import pysparksqlfunctions as F
from pysparksqltypes import LongType
import copy

# 读取parquet文件数代码
df1 sparkreadload(
path＇<存储路径><表名>＇
format＇parquet＇ headerTrue)

# 获取表结构
_schema copydeepcopy(df1schema)
df2 df1rddzipWithIndex()map(lambda l list(l[0]) + [l[1]])toDF(_schema)
subprocesscheck_call(＇rm r <存储路径><表名>＇)shellTrue)

# 写入空数集parquet文件
df2writeparquet(
path＇<存储路径><表名>＇
modeoverwrite)

Hive部表中：
from pysparksql import HiveContext
hive HiveContext(sparksparkContext)
hivesql(＇truncate table [<架构名称>]表名＇)
Hive外部表中：
from pysparksql import HiveContext
hive HiveContext(sparksparkContext)
hivesql(＇insert overwrite table [<架构名称>]表名 select * from [<架构名称>]表名 where 12＇)
复制表结构新表
CREATE TABLE [`<架构名称2>`] `<表名2>` LIKE [`<架构名称1>`] `<表名1>`

通show create table [`<架构名称1>`] `<表名1>`语句旧表创建命令列出需该命令拷贝出更改table名字变成[`<架构名称2>`] `<表名2>`建立完全样表
CREATE TABLE [<架构名称2>]<表名2> AS [<架构名称1>]<表名1> WITH NO DATA

通show table [<架构名称1>] <表名1>语句旧表创建命令列出需该命令拷贝出更改table名字变成[<架构名称2>] <表名2>建立完全样表
Parquet文件中：
import pysparksqlfunctions as F
from pysparksqltypes import LongType
import copy

# 读取parquet文件数代码
df1 sparkreadload(
path＇<存储路径1><表名1>＇
format＇parquet＇ headerTrue)

# 获取表结构
_schema copydeepcopy(df1schema)
df2 df1rddzipWithIndex()map(lambda l list(l[0]) + [l[1]])toDF(_schema)

# 写入空数集parquet文件
df2writeparquet(
path＇<存储路径2><表名2>＇
modeoverwrite)

Hive表中：
CREATE TABLE [<架构名称2>]<表名2> LIKE [<架构名称1>]<表名1>

通desc formmated [<架构名称1>] <表名1>语句show create table [<架构名称1>] <表名1>语句旧表创建命令列出需该命令拷贝出更改表名字变成[<架构名称2>] <表名2>建立完全样表
创建表插入查询数
CREATE TABLE [`<架构名称>`] `<表名>` (
<字段名1> <类型1>[ AUTO_INCREMENT]
<字段名2> <类型2>[ AUTO_INCREMENT]
<字段名3> <类型3>[ AUTO_INCREMENT]
…
<字段名n> <类型3n>[ AUTO_INCREMENT] [
PRIMARY KEY (<键字段名>)][
UNIQUE (<唯值字段名1> <唯值字段名2><唯值字段名3>…<唯值字段名m>)]
) [ENGINE{InnoDB|MYISAM|BDB} DEFAULT CHARSET{utf8|gbk}]

INSERT INTO [`<架构名称>`]`<表名>`

者

CREATE TABLE [`<架构名称>`] `<表名>`

CREATE {MULTISET|SET} TABLE [<架构名称>]<表名>[
<参数1>
<参数2>
<参数3>
…
<参数n>]
(
<字段名1> <类型1> [CHARACTER SET <字符集1> NOT CASESPECIFIC]
<字段名2> <类型2> [CHARACTER SET <字符集2> NOT CASESPECIFIC]
<字段名3> <类型3> [CHARACTER SET <字符集3> NOT CASESPECIFIC]
…
<字段名n> <类型n> [CHARACTER SET <字符集3> NOT CASESPECIFIC]
)
[UNIQUE] [PRIMARY INDEX (<键字段名>)]

INSERT INTO [<架构名称>]<表名>

者

CREATE TABLE [<架构名称>]<表名> AS (

) WITH DATA

者

CREATE TABLE [<架构名称1>]<表名1> AS [<架构名称2>]<表名2> WITH DATA
Parquet文件中：
sparksql(
<查询语句>
)
writeparquet(
path＇<存储路径><表名>＇
modeoverwrite)

Hive表中：
部表
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(
CREATE TABLE [`<库名>`]`<表名>`(
`<字段名1>` <类型1>
`<字段名2>` <类型2>
`<字段名3>` <类型3>
…
`<字段名n>` <类型n>)
[PARTITIONED BY (
`<分区字段1>` <分区字段类型1>
`<分区字段2>` <分区字段类型2>
`<分区字段3>` <分区字段类型3>
…
`<分区字段n>` <分区字段类型n>
)]
ROW FORMAT SERDE
＇orgapachehadoophiveqlioparquetserdeParquetHiveSerDe＇
STORED AS INPUTFORMAT
＇orgapachehadoophiveqlioparquetMapredParquetInputFormat＇
OUTPUTFORMAT
＇orgapachehadoophiveqlioparquetMapredParquetOutputFormat＇
LOCATION
＇hdfs<表名>＇)

hivesql(
CREATE TABLE [`<库名>`]`<表名>`(
`<字段名1>` <类型1>
`<字段名2>` <类型2>
`<字段名3>` <类型3>
…
`<字段名n>` <类型n>)
[PARTITIONED BY (
`<分区字段1>` <分区字段类型1>
`<分区字段2>` <分区字段类型2>
`<分区字段3>` <分区字段类型3>
…
`<分区字段n>` <分区字段类型n>
)]
ROW FORMAT SERDE
＇orgapachehadoophiveqlioparquetserdeParquetHiveSerDe＇
STORED AS PARQUET)

外部表
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(
CREATE EXTERNAL TABLE [IF NOT EXISTS] [`<库名>`]`<表名>`(
`<字段名1>` <类型1>
`<字段名2>` <类型2>
`<字段名3>` <类型3>
…
`<字段名n>` <类型n>)
[PARTITIONED BY (
`<分区字段1>` <分区字段类型1>
`<分区字段2>` <分区字段类型2>
`<分区字段3>` <分区字段类型3>
…
`<分区字段n>` <分区字段类型n>
)]
ROW FORMAT SERDE
＇orgapachehadoophiveqlioparquetserdeParquetHiveSerDe＇
STORED AS INPUTFORMAT
＇orgapachehadoophiveqlioparquetMapredParquetInputFormat＇
OUTPUTFORMAT
＇orgapachehadoophiveqlioparquetMapredParquetOutputFormat＇
LOCATION
＇hdfs<表名>＇)

hivesql(INSERT OVERWRITE TABLE [`<库名>`]`<表名>` <查询语句> )
插入少量数
INSERT INTO [`<架构名称>`]`<表名>` (<字段名1><字段名2><字段名3><字段名n>) VALUES
(<值1><值2><值3>…<值n>)
(<值n+1><值n+2><值n+3>…<值2n>)
(<值2n+1><值2n+2><值2n+3>…<值3n>)
…
(<值mn+1><值mn+2><值mn+3>…<值(m+1)n >)
INSERT INTO [<架构名称>]<表名>
(<字段名1><字段名2><字段名3><字段名n>)
SELECT *
FROM (SELECT *
FROM (SELECT <值1><值2><值3>…<值n>) t
UNION ALL
SELECT *
FROM (SELECT <值n+1><值n+2><值n+3>…<值2n>) t
UNION ALL
SELECT *
FROM (SELECT <值2n+1><值2n+2><值2n+3>…<值3n>) t

UNION ALL
SELECT *
FROM (SELECT <值mn+1><值mn+2><值mn+3>…<值(m+1)n>) t
) tt
PySpark文件中：
<表名>_df sparkcreateDataFrame([(<值1><值2><值3><值n>)(<值n+1><值n+2><值n+3><值2n>)(<值2n+1><值2n+2><值2n+3><值3n>)(<值mn+1><值mn+2><值mn+3><值(m+1)n>)][＇<字段名1>＇＇<字段名2>＇＇<字段名3>＇＇<字段名n>＇])
dfwriteparquet(
path＇<存储路径><表名>[<分区字段><分区值>]＇)
modeoverwrite)

<表名>_df sparkparallelize([(<值1><值2><值3><值n>)(<值n+1><值n+2><值n+3><值2n>) (<值2n+1><值2n+2><值2n+3><值3n>)…(<值mn+1><值mn+2><值mn+3><值(m+1)n>)])toDF([＇<字段名1>＇＇<字段名2>＇＇<字段名3>＇＇<字段名n>＇])
<表名>_dfwriteparquet(
path＇<存储路径><表名>[<分区字段><分区值>]＇)
modeoverwrite)

Hive表中：
INSERT INTO TABLE [<库名>]<表名> [PARTITION (<分区字段> ＇<分区值>＇)]
VALUES (<值1><值2><值3><值n>) (<值n+1><值n+2><值n+3><值2n>) (<值2n+1><值2n+2><值2n+3><值3n>)…(<值mn+1><值mn+2><值mn+3><值(m+1)n>)
限制查询返回行数
SELECT <字段列表>
FROM [`<架构名称1>`]`<表名>`
{INNERLEFTRIGHTFULL} JOIN [`<架构名称2>`]`<维度表名1>`
ON <表连接条件1>
{INNERLEFTRIGHTFULL} JOIN [`<架构名称3>`]`<度量表名1>`
…
ON <表连接条件2>
[WHERE <筛选条件>] LIMIT <限制返回行数>
SELECT TOP <限制返回行数> <字段列表>
FROM [<架构名称1>]<表名1>
{INNERLEFTRIGHTFULL} JOIN [<架构名称2>]<维度表名1>
ON <表连接条件1>
{INNERLEFTRIGHTFULL} JOIN [<架构名称3>]<度量表名1>
…
ON <表连接条件2>
[WHERE <筛选条件>]
sparksql(
SELECT * FROM [<架构名称1>]<表名1>
)
createOrReplaceTempView(<表名1>)

sparksql(
SELECT * FROM [<架构名称2>]<维度表名1>
)cache()
createOrReplaceTempView(<表名2>)

sparksql(
SELECT * FROM [<架构名称3>]<度量表名1>
)
createOrReplaceTempView(<表名3>)
…
sparksql(
SELECT <字段列表>
FROM <表名1>
JOIN <表名2> ON <表连接条件1>
{INNERLEFTRIGHTFULL} JOIN <表名3> ON <表连接条件2>
[WHERE <筛选条件>])limit(<限制返回行数>)
带表连接查询
SELECT <字段列表>
FROM [`<架构名称1>`]`<表名1>`
{INNERLEFTRIGHTFULL} JOIN [`<架构名称2>`]`<维度表名1>`
ON <表连接条件1>
{INNERLEFTRIGHTFULL} JOIN [`<架构名称3>`]`<度量表名1>`
…
ON <表连接条件2>
[WHERE <筛选条件>]
SELECT <字段列表>
FROM [<架构名称1>]<表名1>
{INNERLEFTRIGHTFULL} JOIN [<架构名称2>]<维度表名1>
ON <表连接条件1>
{INNERLEFTRIGHTFULL} JOIN [<架构名称3>]<度量表名1>
…
ON <表连接条件2>
[WHERE <筛选条件>]
sparksql(
SELECT * FROM [<架构名称1>]<表名1>
)
createOrReplaceTempView(<表名1>)

sparksql(
SELECT * FROM [<架构名称2>]<维度表名1>
)cache()
createOrReplaceTempView(<表名2>)

sparksql(
SELECT * FROM [<架构名称3>]<度量表名1>
)
createOrReplaceTempView(<表名3>)
…
sparksql(
SELECT <字段列表>
FROM <表名1>
JOIN <表名2> ON <表连接条件1>
{INNERLEFTRIGHTFULL} JOIN <表名3> ON <表连接条件2>
[WHERE <筛选条件>])
带表连接更新表记录
CREATE TABLE [`<架构名称1>`] `<表名1>` (
<字段名1> <类型1>[ AUTO_INCREMENT]
<字段名2> <类型2>[ AUTO_INCREMENT]
<字段名3> <类型3>[ AUTO_INCREMENT]
…
<字段名n> <类型3n>[ AUTO_INCREMENT] [
PRIMARY KEY (<键字段名>)][
UNIQUE (<唯值字段名1> <唯值字段名2><唯值字段名3>…<唯值字段名m>)]
) [ENGINE{InnoDB|MYISAM|BDB} DEFAULT CHARSET{utf8|gbk}]

INSERT INTO [`<架构名称1>`] `<表名1>`
SELECT <键字段>
<值变字段1>
<值变字段2>
<值变字段3>
…
<值变字段n>
<值改变字段1>
<值改变字段1>
<值改变字段2>
…
<值改变字段n>
FROM [`<架构名称2>`]`<表名2>`
WHERE <筛选条件>

UPDATE <名1>
FROM [`<架构名称1>`]`<表名1>` AS <名1>[`<架构名称3>``<表名3>`] SET
<值改变字段1><改变值1>
<值改变字段2><改变值2>
<值改变字段3><改变值3>
…
<值改变字段n><改变值n>

WHERE <表连接条件>
[AND <筛选条件>]
CREATE [MULTISET] TABLE [<架构名称1>]<表名1>[
<参数1>
<参数2>
<参数3>
…
<参数n>]
(
<字段名1> <类型1> [CHARACTER SET <字符集1> NOT CASESPECIFIC]
<字段名2> <类型2> [CHARACTER SET <字符集2> NOT CASESPECIFIC]
<字段名3> <类型3> [CHARACTER SET <字符集3> NOT CASESPECIFIC]
…
<字段名n> <类型3> [CHARACTER SET <字符集3> NOT CASESPECIFIC]
)
[UNIQUE] [PRIMARY INDEX (<键字段名>)]

INSERT INTO [<架构名称1>]<表名1>
SELECT <键字段>
<值变字段1>
<值变字段2>
<值变字段3>
…
<值变字段n>
<值改变字段1>
<值改变字段1>
<值改变字段2>
…
<值改变字段n>
FROM [<架构名称2>]<表名2>
WHERE <筛选条件>

UPDATE <名1>
FROM [<架构名称1>]<表名1> AS <名1>[<架构名称3><表名3>] SET
<值改变字段1><改变值1>
<值改变字段2><改变值2>
<值改变字段3><改变值3>
…
<值改变字段n><改变值n>

WHERE <表连接条件>
[AND <筛选条件>]
sparksql(
SELECT * FROM <架构名称2><表名2>
)
createOrReplaceTempView(<表名1>)

sparksql(
SELECT * FROM <架构名称3><表名3>
)
createOrReplaceTempView(<表名2>)

sparksql(
SELECT <名1><键字段>
<值变字段1>
<值变字段2>
<值变字段3>
…
<值变字段n>
if(<名2><键字段> is null <名1><值改变字段1> <改变值1>) AS <值改变字段1>
if(<名2><键字段> is null <名1><值改变字段2> <改变值2>) AS <值改变字段2>
if(<名2><键字段> is null <名1><值改变字段3> <改变值3>) AS <值改变字段3>
…
if(<名2><键字段> is null <名1><值改变字段n> <改变值n>) AS <值改变字段n>
FROM <表名1> AS <名1>
INNER JOIN <表名2> AS <名2>
ON <表连接条件>
[WHERE <筛选条件>])
writeparquet(
path＇<存储路径><表名1>＇
modeoverwrite)
合数
REPLACE INTO [`<架构名称>`] `<表名>` (<键字段名> <字段名1> <字段名2> <字段名3> … <字段名n>) VALUES (<键值> <值1> <值2> <值3> … <值n>)

LOAD DATA LOCAL INFILE ＇<存储路径><文件名>＇ REPLACE INTO TABLE [`<架构名称>`] `<表名>`

INSERT INTO [`<架构名称>`] `<表名>` (<键字段名><字段名1> <字段名2> <字段名3> … <字段名n>)
VALUES (<键值1> <值1> <值2> <值3> … <值n>) (<键值2> <值n+1> <值n+2> <值n+3> … <值2n>)
ON DUPLICATE KEY UPDATE <字段名1> VALUES(<字段名1>)<字段名2> VALUES(<字段名2>)<字段名3> VALUES(<字段名3>)…<字段名n> VALUES(<字段名n>)

insert into [`<架构名称>`] `<表名>`(<键字段名><字段名1> <字段名2> <字段名3> … <字段名n>) select * from dupnew on duplicate key update <字段名1> VALUES(<字段名1>)<字段名2> VALUES(<字段名2>)<字段名3> VALUES(<字段名3>)…<字段名n> VALUES(<字段名n>)

insert ignore into [`<架构名称>`] `<表名>`(<键字段名><字段名1> <字段名2> <字段名3> … <字段名n>) values (<键值1> <值1> <值2> <值3> … <值n>)

INSERT IGNORE INTO [`<架构名称>`] `<表名1>` SELECT <键字段名> <字段名1> <字段名2> <字段名3> … <字段名n> FROM [`<架构名称>`] `<表名2>`

SELECT <键字段名> <字段名1> <字段名2> <字段名3> … <字段名n> FROM [`<架构名称>`] `<表名1>` UNION DISTINCT SELECT <键字段名> <字段名1> <字段名2> <字段名3> … <字段名n> FROM [`<架构名称>`] `<表名2>`

创建测试表
drop table test_a
create table test_a(
id VARCHAR (16)
name VARCHAR (16)
Operatime datetime
)
drop table test_b
create table test_b(
id VARCHAR (16)
name VARCHAR (16)
Operatime datetime
)

插入模拟数
INSERT into test_b values(11now())(22now())
INSERT into test_a values(11now())(33now())

查询数
SELECT * FROM test_b
SELECT * FROM test_a

delimiter
CREATE PROCEDURE merge_a_to_b () BEGIN
定义需插入a表插入b表程变量
DECLARE _ID VARCHAR (16)
DECLARE _NAME VARCHAR (16)
游标遍历数结束标志
DECLARE done INT DEFAULT FALSE
游标指a表结果集第条1位置
DECLARE cur_account CURSOR FOR SELECT ID NAME FROM test_a
游标指a表结果集条加1位置设置结束标志
DECLARE CONTINUE HANDLER FOR NOT FOUND SET done TRUE
开游标
OPEN cur_account
遍历游标
read_loop
LOOP
取值a表前位置数时变量
FETCH NEXT FROM cur_account INTO _ID_NAME

果取值结束跳出循环
IF done THEN LEAVE read_loop
END IF

前数做果b表存更新时间存插入
IF NOT EXISTS ( SELECT 1 FROM TEST_B WHERE ID _ID AND NAME_NAME )
THEN
INSERT INTO TEST_B (ID NAMEoperatime) VALUES (_ID_NAMEnow())
ELSE
UPDATE TEST_B set operatime now() WHERE ID _ID AND NAME_NAME
END IF

END LOOP
CLOSE cur_account

END

delimiter
CREATE PROCEDURE merge_a_to_b () BEGIN
定义需插入a表插入b表程变量
DECLARE _ID VARCHAR (16)
DECLARE _NAME VARCHAR (16)
游标遍历数结束标志
DECLARE done INT DEFAULT FALSE
游标指a表结果集第条1位置
DECLARE cur_account CURSOR FOR SELECT ID NAME FROM test_a
游标指a表结果集条加1位置设置结束标志
DECLARE CONTINUE HANDLER FOR NOT FOUND SET done TRUE
开游标
OPEN cur_account
遍历游标
read_loop
LOOP
取值a表前位置数时变量
FETCH NEXT FROM cur_account INTO _ID_NAME

果取值结束跳出循环
IF done THEN LEAVE read_loop
END IF

前数做果b表存更新时间存插入
IF NOT EXISTS ( SELECT 1 FROM TEST_B WHERE ID _ID AND NAME_NAME )
THEN
INSERT INTO TEST_B (ID NAMEoperatime) VALUES (_ID_NAMEnow())
ELSE
UPDATE TEST_B set operatime now() WHERE ID _ID AND NAME_NAME
END IF

END LOOP
CLOSE cur_account

END
merge into [<架构名称1>]<表名1> <名1>
using [<架构名称2>]<表名2> <名2>
on (<名1><连接字段名1> <名2><连接字段名2>)
when matched then
update set <字段名1> <名2><字段名1><字段名2> <名2><字段名3><字段名3> <名2><字段名3>…<字段名n> <名2><字段名n>
when not matched then
insert values(<名2><连接字段名2><名2><字段名3><名2><字段名3>…<字段名n> <名2><字段名n>)
py
<表名1>_df sparkreadload(
path＇<存储路径1><表名1>＇
format＇parquet＇ headerTrue)

<表名2>_df sparkreadload(
path＇<存储路径2><表名2>＇
format＇parquet＇ headerTrue)

<表名1>_dfcreateOrReplaceTempView(<表名1>)
<表名2>_dfcreateOrReplaceTempView(<表名2>)

<表名1>_merge_df sparksql(
SELECT ifnull(ODS<键字段名>STG<键字段名>) AS <键字段名>ifnull(ODS<字段名1>STG<字段名1>) AS <字段名1>ifnull(ODS<字段名2>STG<字段名2>) AS <字段名2>ifnull(ODS<字段名3>STG<字段名3>) AS <字段名3> …ifnull(ODS<字段名n>STG<字段名n>) AS <字段名n> FROM
(
SELECT <键字段名> <字段名1> <字段名2> <字段名3> … <字段名n>
FROM <表名2>
) STG
FULL JOIN <表名1> AS ODS ON STG<键字段名> ODS<键字段名>
)

<表名1>_merge_dfwriteparquet(
path＇<存储路径1><表名1>_merge＇
modeoverwrite)

py
<表名1>_merge_df sparkreadload(
path＇<存储路径1><表名1>_merge＇
format＇parquet＇ headerTrue)

<表名1>_merge_dfwriteparquet(
path＇<存储路径1><表名1>＇
modeoverwrite)

Hive表中：
CREATE TABLE [`<库名>`]`<表名>` (
`<字段名1>` <类型1>
`<字段名2>` <类型2>
`<字段名3>` <类型3>
…
`<字段名n>` <类型n>
)
CLUSTERED BY (<键字段>) INTO <数字> buckets
ROW FORMAT DELIMITED FIELDS TERMINATED BY ＇＇
STORED AS orc
TBLPROPERTIES(＇transactional＇＇true＇)

MERGE INTO [`<库名>`]`<表名>` AS <名1>
USING (
<查询语句>
) AS <名2>
ON …
WHEN MATCHED
THEN
UPDATE
SET <字段名1> <值1>
<字段名2> <值2>
<字段名3> <值3>
…
<字段名n> <值n>
查询分组排名数
SELECT <字段1>
<字段2>
<字段3>
…
<字段n>
FROM (
SELECT <字段1>
<字段2>
<字段3>
…
<字段n>
ROW_NUMBER() OVER (
PARTITION BY <分组字段> ORDER BY <排序字段> [DESC]
) AS rn
FROM [`<架构名称>`] ` <表名> `
[WHERE <筛选条件>]
) t
WHERE rn 1
SELECT <字段1>
<字段2>
<字段3>
…
<字段n>
FROM [<架构名称>]<表名>
QUALIFY ROW_NUMBER() OVER(PARTITION BY <分组字段> ORDER BY <排序字段> [DESC]) 1
[WHERE <筛选条件>]
<表名>_df sparksql(
SELECT * FROM [<架构名称>]<表名>
)
<表名>_dfcreateOrReplaceTempView(<表名>)
<表名>_unique_df sparksql(
SELECT <字段1>
<字段2>
<字段3>
…
<字段n>
FROM (
SELECT <字段1>
<字段2>
<字段3>
…
<字段n>
ROW_NUMBER() OVER (
PARTITION BY <分组字段> ORDER BY <排序字段> [DESC]
) AS rn
FROM <表名>
[WHERE <筛选条件>]
) t
WHERE rn 1)

Hive表中：
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(
SELECT
<字段1>
<字段2>
<字段3>
…
<字段n>
FROM (
SELECT
<字段1>
<字段2>
<字段3>
…
<字段n>
ROW_NUMBER() OVER (
PARTITION BY <分组字段> ORDER BY <排序字段>
) AS rn
FROM [`<库名>`]`<表名>`
[WHERE <筛选条件>]
) t
WHERE rn 1)
字符串连接
SELECT CONCAT(<字符串变量字段常量1><字符串变量字段常量2>)
SELECT <字符串变量字段常量1> || <字符串变量字段常量2>
sparksql(
SELECT CONCAT(<字符串变量字段常量1><字符串变量字段常量2>))
查询分组里数字
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<聚合函数1>(<度量字段1>) AS <名1>
<聚合函数2>(<度量字段2>) AS <名2>

<聚合函数3>(<度量字段3>) AS <名3>
…
<聚合函数m>(<度量字段m>) AS <名m>
FROM [<架构名称>]<表名>
GROUP BY 123…n
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<聚合函数1>(<度量字段1>) AS <名1>
<聚合函数2>(<度量字段2>) AS <名2>

<聚合函数3>(<度量字段3>) AS <名3>
…
<聚合函数m>(<度量字段m>) AS <名m>
FROM [<架构名称>]<表名>
GROUP BY 123…n
sparksql(
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<聚合函数1>(<度量字段1>) AS <名1>
<聚合函数2>(<度量字段2>) AS <名2>

<聚合函数3>(<度量字段3>) AS <名3>
…
<聚合函数m>(<度量字段m>) AS <名m>
FROM [<架构名称>]<表名>
GROUP BY <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>)

DECIMAL类型转换
SELECT (CAST(<数值字段变量常量1> AS DECIMAL(382)) CAST(<数值字段变量常量2> AS DECIMAL(382)))
SELECT (CAST(<数值字段变量常量1> AS DECIMAL(382)) CAST(<数值字段变量常量2> AS DECIMAL(382)))
<变量> sparksql(
SELECT <数值字段变量常量1> * 100 <数值字段变量常量2> 100)
NULL值换
IFNULL(exp1exp2)

COALESCE(exp1exp2)
字段exp1NULL值时返回exp1否返回exp2
NULLIF(exp1exp2)

COALESCE(exp1exp2)
字段exp1NULL值时返回exp1否返回exp2
>>> df sparkcreateDataFrame([(1) (2) (3) (None)] [＇col＇])
>>> dfshow()
++
| col|
++
| 1|
| 2|
| 3|
|null|
++
>>> df dffillna({＇col＇＇4＇})
>>> dfshow()
or dffillna({＇col＇＇4＇})show()
++
|col|
++
| 1|
| 2|
| 3|
| 4|
++

sparksql(
SELECT IFNULL(exp1exp2) …
)
获取年月日获取中国时区天日期
SELECT YEAR(CURRENT_DATE()) MONTH(CURRENT_DATE()) DAY(CURRENT_DATE()) CONVERT_TZ(create_time @@sessiontime_zone＇+800＇)
SET time_zone＇AsiaShanghai＇
select now()
SELECT EXTRACT(YEAR FROM CURRENT_DATE) EXTRACT(MONTH FROM CURRENT_DATE) EXTRACT(DAY FROM CURRENT_DATE) CAST(CONVERT_TIMEZONE(＇AsiaShanghai＇CAST(GETDATE() AS TIMESTAMP)) AS DATE)
<变量> sparksql(
SELECT YEAR(CURRENT_DATE) MONTH(CURRENT_DATE) DAY(CURRENT_DATE) CAST(CONVERT_TIMEZONE(＇AsiaShanghai＇CAST(GETDATE() AS TIMESTAMP)) AS DATE))
时间戳间间隔天数计算
SELECT TIMESTAMPDIFF(DAY <开始时间戳> <结束时间戳>)
SELECT EXTRACT(DAY FROM (<结束时间戳> <开始时间戳> DAY(4) TO SECOND)) * 86400
from pysparksqlfunctions import *

dates [(120190701 120119111)
(220190624 120119222)
(320191116 164455406)
(420191116 165059406)
]

df sparkcreateDataFrame(datadates schema[idinput_timestamp])

#Calculate Time difference in Seconds
<变量>dfwithColumn(＇from_timestamp＇to_timestamp(col(＇from_timestamp＇)))＼
withColumn(＇end_timestamp＇ current_timestamp())＼
withColumn(＇DiffInDays＇(col(end_timestamp)cast(long) col(＇from_timestamp＇)cast(long))246060)
<变量>show(truncateFalse)
列出表字段信息
查系统表
SELECT
ColumnId 字段键（建表序致）
DataBaseName属库
TableName表名
DefaultValue默认值
ColumnName字段名
ColumnTitle字段名
ColumnType字段类型
ColumnLength字段长度
DecimalTotalDigits精度
DecimalFractionalDigits 标度
ColumnFormat 格式
FROM
DBCColumns
WHERE
DATABASENAME＇<库名>＇
AND TABLENAME＇<表名>＇
ORDER BY 1
查表结构
SHOW TABLE [`<库名>`] `<表名>`
字段类型映射关系

字段类型
映射值
拼接规
CF
CHAR
a ASCII编码(LATIN) CHAR(长度)
b UNICODE编码 CHAR(长度2)
CV
VARCHAR
a ASCII编码(LATIN) VARCHAR(长度)
b UNICODE编码 VARCHAR(长度2)
D
DECIMAL
DECIMAL(精度标度)
DA
DATE
DATE FORMAT 格式’
I
INTEGER
INTEGER
I8
BIGINT
BIGINT

SELECT

COLUMN_NAME AS ＇字段名＇

DATA_TYPE AS `数类型`

CHARACTER_MAXIMUM_LENGTH AS `字符长度`

NUMERIC_PRECISION AS `数字长度`

NUMERIC_SCALE AS `数位数`

IS_NULLABLE AS `否允许非空`

CASE WHEN EXTRA ＇auto_increment＇ THEN 1 ELSE 0 END AS `否增`

COLUMN_DEFAULT AS `默认值`

COLUMN_COMMENT AS `备注`

FROM information_schemaCOLUMNS WHERE TABLE_SCHEMA＇<库名>＇ AND TABLE_NAME＇<表名>＇
df…

dfschema

dfprintSchema()

for name dtype in dfdtypes
print(name dtype)
分区操作
查MySQL否支持分区
1MySQL56前版
show variables like ＇partition＇

2MySQL57
show plugins

二分区表分类限制
1分区表分类
RANGE分区：基属定连续区间列值行分配分区

LIST分区：类似RANGE分区区LIST分区基列值匹配离散值集合中某值进行选择

HASH分区：基户定义表达式返回值进行选择分区该表达式插入表中行列值进行计算函数包含MySQL 中效产生非负整数值表达式

KEY分区：类似HASH分区区KEY分区支持计算列列MySQL服务器提供身哈希函数必须列列包含整数值

复合分区：MySQL 56版中支持RANGELIST子分区子分区类型HASHKEY

2分区表限制
1）分区键必须包含表键唯键中

2）MYSQL分区函数列身进行较时滤分区根表达式值滤分区表达式分区函数行

3）分区数： NDB存储引擎定表分区数8192（包括子分区）果分区数未达8192时提示 Got error … from storage engine Out of resources when opening file通增加open_files_limit系统变量值解决问题然时开文件数量操作系统限制

4）支持查询缓存：分区表支持查询缓存涉分区表查询动禁查询缓存法启类查询

5）分区innodb表支持外键

6）服务器SQL_mode影响分区表步复制机机SQL_mode会导致sql语句导致分区间数分配定位置甚导致插入机成功分区表库失败获佳效果您应该始终机机相服务器SQL模式

7）ALTER TABLE … ORDER BY：分区表运行ALTER TABLE … ORDER BY列语句会导致分区中行排序

8）全文索引分区表支持全文索引InnoDBMyISAM存储引擎分区表
9）分区表法外键约束
10）Spatial columns：具空间数类型（POINTGEOMETRY）列分区表中
11）时表：时表分区
12）subpartition问题： subpartition必须HASHKEY分区 RANGELIST分区分区 HASHKEY分区子分区
13）分区表支持mysqlcheckmyisamchkmyisampack

三创建分区表
1range分区
行数基定连续区间列值放入分区
CREATE TABLE `test_11` (
`id` int(11) NOT NULL
`t` date NOT NULL
PRIMARY KEY (`id``t`)
) ENGINEInnoDB DEFAULT CHARSETutf8
PARTITION BY RANGE (to_days(t))
(PARTITION p20170801 VALUES LESS THAN (736907) ENGINE InnoDB
PARTITION p20170901 VALUES LESS THAN (736938) ENGINE InnoDB
PARTITION pmax VALUES LESS THAN maxvalue ENGINE InnoDB)123456789
然插入4条数：
insert into test_11 values (120170722)(220170822)(320170823)(420170824)1
然查informationpartitions分区信息统计：
select PARTITION_NAME as 分区TABLE_ROWS as 行数 from information_schemapartitions where table_schemamysql_test and table_nametest_11
+++
| 分区 | 行数 |
+++
| p20170801 | 1 |
| p20170901 | 3 |
+++
2 rows in set (000 sec)12345678
出分区p20170801插入1行数p20170901插入3行数
yearto_daysunix_timestamp等函数相应时间字段进行转换然分区
2list分区
range分区样list分区面离散值
mysql> CREATE TABLE h2 (
> c1 INT
> c2 INT
> )
> PARTITION BY LIST(c1) (
> PARTITION p0 VALUES IN (1 4 7)
> PARTITION p1 VALUES IN (2 5 8)
> )
Query OK 0 rows affected (011 sec)123456789
RANGE分区情况没catchallMAXVALUE 分区表达式预期值应PARTITION … VALUES IN（…）子句中涵盖包含匹配分区列值INSERT语句失败显示错误示例示：
mysql> INSERT INTO h2 VALUES (3 5)
ERROR 1525 (HY000) Table has no partition for value 312
3hash分区
根户定义表达式返回值进行分区返回值负数
CREATE TABLE t1 (col1 INT col2 CHAR(5) col3 DATE)
PARTITION BY HASH( YEAR(col3) )
PARTITIONS 4123
果插入col3数值’20050915’根计算选择插入分区：
MOD(YEAR(＇20050901＇)4)
MOD(20054)
1123
4key分区
根MySQL数库提供散列函数进行分区
CREATE TABLE k1 (
id INT NOT NULL
name VARCHAR(20)
UNIQUE KEY (id)
)
PARTITION BY KEY()
PARTITIONS 21234567
KEY仅列出零列名称作分区键列必须包含表键部分全部果该表具果没列名称作分区键表键（果）果没键唯键唯键分区键果唯键列未定义NOT NULL条语句失败
分区类型KEY分区限整数空值例CREATE TABLE语句效：
CREATE TABLE tm1 (
s1 CHAR(32) PRIMARY KEY
)
PARTITION BY KEY(s1)
PARTITIONS 1012345
注意：key分区表执行ALTER TABLE DROP PRIMARY KEY样做会生成错误 ERROR 1466 (HY000) Field in list of fields for partition function not found in table
5Column分区
COLUMN分区55开始引入分区功RANGE COLUMNLIST COLUMN两种分区支持整形日期字符串RANGELIST分区方式非常相似
COLUMNSRANGELIST分区区
1）针日期字段分区需函数进行转换例针date字段进行分区需YEAR()表达式进行转换
2）COLUMN分区支持字段作分区键支持表达式作分区键
column支持数类型：
1）整型floatdecimal支持
2）日期类型：datedatetime支持
3）字符类型：CHAR VARCHAR BINARYVARBINARYblobtext支持
单列column range分区mysql> show create table list_c
CREATE TABLE `list_c` (
`c1` int(11) DEFAULT NULL
`c2` int(11) DEFAULT NULL
) ENGINEInnoDB DEFAULT CHARSETlatin1
*50500 PARTITION BY RANGE COLUMNS(c1)
(PARTITION p0 VALUES LESS THAN (5) ENGINE InnoDB
PARTITION p1 VALUES LESS THAN (10) ENGINE InnoDB) *
列column range分区mysql> show create table list_c
CREATE TABLE `list_c` (
`c1` int(11) DEFAULT NULL
`c2` int(11) DEFAULT NULL
`c3` char(20) DEFAULT NULL
) ENGINEInnoDB DEFAULT CHARSETlatin1
*50500 PARTITION BY RANGE COLUMNS(c1c3)
(PARTITION p0 VALUES LESS THAN (5＇aaa＇) ENGINE InnoDB
PARTITION p1 VALUES LESS THAN (10＇bbb＇) ENGINE InnoDB) *
单列column list分区mysql> show create table list_c
CREATE TABLE `list_c` (
`c1` int(11) DEFAULT NULL
`c2` int(11) DEFAULT NULL
`c3` char(20) DEFAULT NULL
) ENGINEInnoDB DEFAULT CHARSETlatin1
*50500 PARTITION BY LIST COLUMNS(c3)
(PARTITION p0 VALUES IN (＇aaa＇) ENGINE InnoDB
PARTITION p1 VALUES IN (＇bbb＇) ENGINE InnoDB) *
6子分区（组合分区）
分区基础进步分区时成复合分区
MySQL数库允许rangelist分区进行HASHKEY子分区例：
CREATE TABLE ts (id INT purchased DATE)
PARTITION BY RANGE( YEAR(purchased) )
SUBPARTITION BY HASH( TO_DAYS(purchased) )
SUBPARTITIONS 2 (
PARTITION p0 VALUES LESS THAN (1990)
PARTITION p1 VALUES LESS THAN (2000)
PARTITION p2 VALUES LESS THAN MAXVALUE
)
[root@mycat3 ~]# ll datamysql_data_3306mysql_testts*
rwr 1 mysql mysql 8596 Aug 8 1354 datamysql_data_3306mysql_testtsfrm
rwr 1 mysql mysql 98304 Aug 8 1354 datamysql_data_3306mysql_testts#P#p0#SP#p0sp0ibd
rwr 1 mysql mysql 98304 Aug 8 1354 datamysql_data_3306mysql_testts#P#p0#SP#p0sp1ibd
rwr 1 mysql mysql 98304 Aug 8 1354 datamysql_data_3306mysql_testts#P#p1#SP#p1sp0ibd
rwr 1 mysql mysql 98304 Aug 8 1354 datamysql_data_3306mysql_testts#P#p1#SP#p1sp1ibd
rwr 1 mysql mysql 98304 Aug 8 1354 datamysql_data_3306mysql_testts#P#p2#SP#p2sp0ibd
rwr 1 mysql mysql 98304 Aug 8 1354 datamysql_data_3306mysql_testts#P#p2#SP#p2sp1ibd
1234567891011121314151617
ts表根purchased进行range分区然进行次hash分区形成3*2分区物理文件证实分区方式通subpartition语法显示指定子分区名称
注意：子分区数量必须相果分区表子分区已subpartition必须表明子分区名称subpartition子句必须包括子分区名字子分区名字必须致
外MyISAM表index directorydata direactory指定分区数索引目录innodb表说该存储引擎表空间动进行数索引理会忽略指定indexdata语法

四普通表转换分区表
1alter table table_name partition by命令重建分区表

alter table jxfp_data_bak PARTITION BY KEY(SH) PARTITIONS 8

五分区表操作
CREATE TABLE t1 (
id INT
year_col INT
)
PARTITION BY RANGE (year_col) (
PARTITION p0 VALUES LESS THAN (1991)
PARTITION p1 VALUES LESS THAN (1995)
PARTITION p2 VALUES LESS THAN (1999)
)

1ADD PARTITION （新增分区）
ALTER TABLE t1 ADD PARTITION (PARTITION p3 VALUES LESS THAN (2002))

2DROP PARTITION （删分区）
ALTER TABLE t1 DROP PARTITION p0 p1

3TRUNCATE PARTITION（截取分区）
ALTER TABLE t1 TRUNCATE PARTITION p0

ALTER TABLE t1 TRUNCATE PARTITION p1 p3

4COALESCE PARTITION（合分区）
CREATE TABLE t2 (
name VARCHAR (30)
started DATE
)
PARTITION BY HASH( YEAR(started) )
PARTITIONS 6

ALTER TABLE t2 COALESCE PARTITION 2

5REORGANIZE PARTITION（拆分重组分区）
1）拆分分区

ALTER TABLE table ALGORITHMINPLACE REORGANIZE PARTITION

ALTER TABLE employees ADD PARTITION (
PARTITION p5 VALUES LESS THAN (2010)
PARTITION p6 VALUES LESS THAN MAXVALUE
)

2）重组分区

ALTER TABLE members REORGANIZE PARTITION s0s1 INTO (
PARTITION p0 VALUES LESS THAN (1970)
)
ALTER TABLE tbl_name
REORGANIZE PARTITION partition_list
INTO (partition_definitions)
ALTER TABLE members REORGANIZE PARTITION p0p1p2p3 INTO (
PARTITION m0 VALUES LESS THAN (1980)
PARTITION m1 VALUES LESS THAN (2000)
)
ALTER TABLE tt ADD PARTITION (PARTITION np VALUES IN (4 8))
ALTER TABLE tt REORGANIZE PARTITION p1np INTO (
PARTITION p1 VALUES IN (6 18)
PARTITION np VALUES in (4 8 12)
)

6ANALYZE CHECK PARTITION（分析检查分区）
1）ANALYZE 读取存储分区中值分布情况

ALTER TABLE t1 ANALYZE PARTITION p1 ANALYZE PARTITION p2

ALTER TABLE t1 ANALYZE PARTITION p1 p2

2）CHECK 检查分区否存错误

ALTER TABLE t1 ANALYZE PARTITION p1 CHECK PARTITION p2

7REPAIR分区
修复破坏分区

ALTER TABLE t1 REPAIR PARTITION p0p1

8OPTIMIZE
该命令回收空闲空间分区碎片整理分区执行该命令相次分区执行 CHECK PARTITION ANALYZE PARTITIONREPAIR PARTITION命令

譬

ALTER TABLE t1 OPTIMIZE PARTITION p0 p1

9REBUILD分区
重建分区相先删分区中数然重新插入分区碎片整理

ALTER TABLE t1 REBUILD PARTITION p0 p1

10EXCHANGE PARTITION（分区交换）
分区交换语法

ALTER TABLE pt EXCHANGE PARTITION p WITH TABLE nt

中pt分区表ppt分区(注子分区)nt目标表

实分区交换限制蛮

1） nt分区表

2）nt时表

3）ntpt结构必须致

4）nt存外键约束键外键

5）nt中数位p分区范围外

具体参考MySQL官方文档

11迁移分区（DISCARD IMPORT ）
ALTER TABLE t1 DISCARD PARTITION p2 p3 TABLESPACE

ALTER TABLE t1 IMPORT PARTITION p2 p3 TABLESPACE

实验环境：（mysql57）
源库：1921682200 mysql5716 zhangdbemp_2分区表
目标库：1921682100 mysql5718 test （zhangdbemp表导入目标库test schema）
：源数库中创建测试分区表emp_2然导入数
MySQL [zhangdb]> CREATE TABLE emp_2(
id BIGINT unsigned NOT NULL AUTO_INCREMENT
x VARCHAR(500) NOT NULL
y VARCHAR(500) NOT NULL
PRIMARY KEY(id)
)
PARTITION BY RANGE COLUMNS(id)
(
PARTITION p1 VALUES LESS THAN (1000)
PARTITION p2 VALUES LESS THAN (2000)
PARTITION p3 VALUES LESS THAN (3000)
)
（接着创建存储程导入测试数）
DELIMITER
CREATE PROCEDURE insert_batch()
begin
DECLARE num INT
SET num1
WHILE num < 3000 DO
IF (num100000) THEN
COMMIT
END IF
INSERT INTO emp_2 VALUES(NULL REPEAT(＇X＇ 500) REPEAT(＇Y＇ 500))
SET numnum+1
END WHILE
COMMIT
END
DELIMITER
mysql> select TABLE_NAMEPARTITION_NAME from information_schemapartitions where table_schema＇zhangdb＇
+++
| TABLE_NAME | PARTITION_NAME |
+++
| emp | NULL |
| emp_2 | p1 |
| emp_2 | p2 |
| emp_2 | p3 |
+++
4 rows in set (000 sec)
mysql> select count(*) from emp_2 partition (p1)
++
| count(*) |
++
| 999 |
++
1 row in set (000 sec)
mysql> select count(*) from emp_2 partition (p2)
++
| count(*) |
++
| 1000 |
++
1 row in set (000 sec)
mysql> select count(*) from emp_2 partition (p3)
++
| count(*) |
++
| 1000 |
++
1 row in set (000 sec)
面出emp_2分区表已创建完成3子分区分区点数
：目标数库中创建emp_2表结构数（源库show create table emp_2＼G 方法查创建该表sql）
MySQL [test]> CREATE TABLE `emp_2` (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT
`x` varchar(500) NOT NULL
`y` varchar(500) NOT NULL
PRIMARY KEY (`id`)
) ENGINEInnoDB AUTO_INCREMENT3000 DEFAULT CHARSETutf8mb4
*50500 PARTITION BY RANGE COLUMNS(id)
(PARTITION p1 VALUES LESS THAN (1000) ENGINE InnoDB
PARTITION p2 VALUES LESS THAN (2000) ENGINE InnoDB
PARTITION p3 VALUES LESS THAN (3000) ENGINE InnoDB) *
[root@localhost test]# ll
rwr 1 mysql mysql 98304 May 25 1558 emp_2#P#p0ibd
rwr 1 mysql mysql 98304 May 25 1558 emp_2#P#p1ibd
rwr 1 mysql mysql 98304 May 25 1558 emp_2#P#p2ibd
注意：
※约束条件字符集等等必须致建议show create table t1 获取创建表SQL否新服务器导入表空间时候会提示1808错误
：目标数库丢弃分区表表空间
MySQL [test]> alter table emp_2 discard tablespace
Query OK 0 rows affected (012 sec)
[root@localhost test]# ll 时候刚3分区idb文件没
rwr 1 mysql mysql 8604 May 25 0414 emp_2frm
：源数库运行FLUSH TABLES … FOR EXPORT 锁定表生成cfg元数文件cfgibd文件传输目标数库中
mysql> flush tables emp_2 for export
Query OK 0 rows affected (000 sec)
[root@localhost zhangdb]# scp emp_2* root@1921682100mysqldatatest 文件cp目标数库
mysql> unlock tables 表锁否
：目标数库中文件授权然导入表空间查数否完整
[root@localhost test]# chown mysqlmysql emp_2#*
MySQL [test]> alter table emp_2 import tablespace
Query OK 0 rows affected (096 sec)
MySQL [test]> select count(*) from emp_2
++
| count(*) |
++
| 2999 |
++
1 row in set (063 sec)
面查知分区表已导入目标数库中
外部分子分区导入目标数库中（整分区表会需子分区导入目标数库中）
部分子分区导入目标数库方法：
1）创建目标表时候需创建导入分区：创建p2 p3两分区
CREATE TABLE `emp_2` (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT
`x` varchar(500) NOT NULL
`y` varchar(500) NOT NULL
PRIMARY KEY (`id`)
) ENGINEInnoDB AUTO_INCREMENT3000 DEFAULT CHARSETutf8mb4
*50500 PARTITION BY RANGE COLUMNS(id)
(
PARTITION p2 VALUES LESS THAN (2000) ENGINE InnoDB
PARTITION p3 VALUES LESS THAN (3000) ENGINE InnoDB) *
2）源库cp目标库文件然俩需分区
3）操作方法样

六获取分区相关信息
1 通 SHOW CREATE TABLE 语句查分区表分区子句
譬mysql> show create table eG

2 通 SHOW TABLE STATUS 语句查表否分区应Create_options字段
譬
mysql> show table statusG

*************************** 1 row ***************************

Name e Engine InnoDB Version 10 Row_format Compact Rows 6 Avg_row_length 10922 Data_length 65536Max_data_length 0 Index_length 0 Data_free 0 Auto_increment NULL Create_time 20151207 222606 Update_time NULL Check_time NULL Collation latin1_swedish_ci Checksum NULL Create_options partitioned Comment

3 查 INFORMATION_SCHEMAPARTITIONS表
4 通 EXPLAIN PARTITIONS SELECT 语句查具体SELECT语句会访问分区

七MySQL57partition表改进
HANDLER statements：MySQL 571分区表开始支持HANDLER语句
index condition pushdown：MySQL573分区表开始支持ICP
load data：MySQL57开始缓存实现性提升分区130KB缓区实现点
Perpartition索引缓存：MySQL57开始支持CACHE INDEXLOAD INDEX INTO CACHE语句分区MyISAM表支持索引缓存
FOR EXPORT选项（FLUSH TABLES）MySQL 574分区InnoDB表开始支持FLUSH TABLES语句FOR EXPORT选项
MySQL 572开始子分区支持ANALYZECHECKOPTIMIZEREPAIRTRUNCATE操作
Teradata分区中常时间分区例添加create table语句末尾实现2013年全年天分区（省事次分510年）：

PARTITION BY RANGE_N(

Rcd_Dt BETWEEN DATE ＇20130101＇ AND DATE ＇20131231＇

EACH INTERVAL ＇1＇ DAY NO RANGE

)

外常（容易掌握）字符串取值分区述时间分区中RANGE_N关键字值分区采CASE_N关键字例示：

PARTITION BY CASE_N(

(CASE WHEN (my_field＇A＇) THEN (1) ELSE (0) END)1

(CASE WHEN (my_field＇B＇) THEN (2) ELSE (0) END)2

(CASE WHEN (my_field＇C＇) THEN (3) ELSE (0) END)3

NO CASE OR UNKNOWN)

更进步中面语法元素：

my_field＇A＇

修改类似样形式：

SUBSTR(my_field11) IN (＇E＇＇F＇＇G＇)

现实中访问数全表扫描变成分区扫描原某步骤达成10100倍性提升复杂耗时较长作业总够缩短半运行时间

1数分区

分布式程序中通信代价较通数集节点间分区进行控制获较少网络传输提升整体性果定RDD需扫描次完全没必预先进行处理数集次诸连接种基键操作中时分区会帮助

Spark法显示控制键具体落工作节点Spark确保组键出现节点

Join操作例果未根RDD中键重新分默认情况连接操作会两数集中键哈希值求出哈希值相记录通网络传输台机器然台机器键相记录进行连接操作

2partitionBy()算子
from pyspark import SparkContextSparkConf
if __name__ ＇__main__＇
conf SparkConf()setMaster(local)setAppName(word_count)
scSparkContext(confconf)
pair_rddscparallelize([(＇a＇1)(＇b＇10)(＇c＇4)(＇d＇7)(＇e＇9)(＇f＇10)])
rdd_1pair_rddpartitionBy(2partitionFunclambda xord(x)2)persist()
print(rdd_1glom()collect())

结果：
rdd_1
[[(＇b＇ 10) (＇d＇ 7) (＇f＇ 10)] [(＇a＇ 1) (＇c＇ 4) (＇e＇ 9)]]
parittionBy()pairRDDpairRDD中key传入partitionFunc函数中需注意果partitonBy（）操作结果持久化面次RDD时会重复数进行分区操作样话partitionBy()重新分区带处会抵消通算子完成python中定义分区scala语言中样麻烦（spark身提供HashPartitioner RangePartitioner）

3影响分区方式操作

算子会生成结果RDD设分区方式：

cogroup()groupWith()join()leftOuterJoin（）rightOuterJoin()groupByKey()reduceByKey()combineByKey()partitionBy()sort()mapValues(果父RDD分区方式）flatMapValues(果父RDD分区方式）filter(果父RDD分区方式）

写入分区
<变量> sparksql(
<查询语句>
)
<变量>writeparquet(
path＇<存储路径><表名>{par_col}{par_val}＇format(par_col＇<分区列名>＇par_val＇<分区值>＇)
modeoverwrite)

删分区
Parquet文件中：
import subprocess

subprocesscheck_call(＇rm r <存储路径><表名>{par_col}{par_val}＇format(par_col＇<分区列名>＇par_val＇<分区值>＇)shellTrue)

Hive表中：
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(＇alter table [`<库名>`]`<表名>` drop [if exists] partition(<分区列名>＇<分区值>＇)＇)

查询分区：
Hive表中：
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(＇show partitions `[<库名>]<表名>`＇)

显示分区HDFS存储路径
Parquet文件中：
import subprocess

subprocesscheck_call(＇hdfs dfs ls hdfs<表名>＇)shellTrue)

Hive表中：
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(
DESC FORMATTED [`<库名>`]`<表名>` partition (<分区列名>＇<分区值>＇))

hivesql(
DESC EXTENDED [`<库名>`]`<表名>` partition (<分区列名>＇<分区值>＇))

hivesql(
USE `<库名>`
SHOW TABLE EXTENDED LIKE `<表名>` PARTITION(<分区列名>＇<分区值>＇))

添加分区加载分区数：
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(
alter table [`<库名>`]`<表名>` add partition (<分区列名>＇<分区值>＇) location ＇hdfs<表名>＇) #改变源数存储位置
hivesql(
load data inpath ＇<存储路径><表名>[<分区列名><分区值>]＇ [overwrite] into table [`<库名>`]`<表名>` partition(<分区列名>＇<分区值>＇)) #会源数切hive表指定路径

修改表分区名称
ALTER TABLE [`<库名>`]`<表名>` PARTITION ((<分区列名>＇<分区值1>＇) RENAME TO PARTITION ((<分区列名>＇<分区值2>＇)

修复Hive表元数分区（般加表创建语句分区操作语句面）
from pysparksql import HiveContext

hive HiveContext(sparksparkContext)
hivesql(
MSCK REPAIR TABLE [`<库名>`]`<表名>`)

Shell命令提交运行脚
vim [<存储路径>]<文件名>sql

mysql [h ] [u <账号>] [p<密码>] [<数库名>] < [<存储路径>]<文件名>sql

mysql [h ] [u <账号>] [p<密码>]
mysql> use <数库名>
mysql> source [<存储路径>]<文件名>sql
mysql> QUIT
文件容：
#binbash
usrbinbteq
LOGON <账号><密码>
<查询语句数操作语句1>
<查询语句数操作语句2>
<查询语句数操作语句3>
…
<查询语句数操作语句n>
IF ERRORCODE <> 0 THEN QUIT ERRORCODE
LOGOFF
QUIT

运行方式：
sh [<存储路径>]<文件名>sh
nohup sh <文件名>sh & #台运行登录断开中断运行
sh <文件名>sh #前台运行登录断开中断运行

文件容：
vim [<存储路径>]<文件名>sh
script_path<脚存储路径>
current_data(date +Ymd)
current_time(date +HMS)
scripts_sql{script_path}sql
sql_file_content`cat {scripts_sql}1`
log_path{scripts_sql}log
log_file{log_path}1{current_time}log
tdserver
dbuser<账号>
dbpass<密码>
dbinfotdserverdbuserdbpass

bteq << END >> log_file 2>&1
LOGON dbinfo
MAXERROR 1

sql_file_content

LOGOFF
END

if [[ ne 0 ]] then
echo e Betq executed 1 failed ＼n Please check the log in ＼n {log_file}
fi

RETCODE
if [ {RETCODE} 0 ]then
echo Please check the error log file log_file
exit 1
else
echo Query executed successfully
fi

cat log_file

运行方式：
nohup sh <文件名>sh sql & #台运行登录断开中断运行
sh <文件名>sh sql #前台运行登录断开中断运行
sparkenvsh
vim HADOOP_HOMEconfsparkenvsh

vim [<存储路径>]sh
SCRIPT_PATH`dirname 0`

SCRIPT_FILENAME`basename 0`
SCRIPT_PATH`dirname 0`
SCRIPT_NAME{SCRIPT_FILENAME*}
SCRIPT_FULL_PATH(readlink f 0)
SCRIPT_ROOT_DIR(dirname {SCRIPT_FULL_PATH})

SCALA_VERSION
SCALA_PATH
SCALA_HOME{SCALA_PATH}scala{SCALA_VERSION}
SPARK_HOME{SPARK_HOME}
SPARK_CONF_DIR
HADOOP_CONF_DIRHADOOP_HOMEetchadoop
HADOOP_VERSION
HBASE_CONF_DIR
STAGE_DIRhdfs
QUEUE{QUEUE}
JAR_DIR{JAR_DIRSCRIPT_ROOT_DIRlib}
LOG_DIRSCRIPT_ROOT_DIRlogs
CONF_DIR{CONF_DIRSCRIPT_ROOT_DIRconf}
EMAIL_SERVER{EMAIL_SERVER<邮件服务器域名IP>}
EMAIL_FROM{EMAIL_FROM<发件邮箱址>}
EMAIL_TO{EMAIL_TO<收件邮箱址>}
TODAY`date +Ymd`
THIS_HOUR`date +HM`
YEAR`date d {TODAY} +Y`
MONTH`date d {TODAY} +m`
DAY`date d {TODAY} +d`
ALERT_EMAILS<通知邮件收件邮箱>
SPARK_DRIVER_CORE{SPARK_DRIVER_CORE2}
SPARK_DRIVER_MEMORY{SPARK_DRIVER_MEMORY8G}
SPARK_EXECUTOR_MEMORY{SPARK_EXECUTOR_MEMORY16G}
SPARK_EXECUTOR_CORE{SPARK_EXECUTOR_CORE2}
SPARK_DEFAULT_PARALLELISM{SPARK_EXECUTOR_CORE150}
SPARK_YARN_TAGS{SPARK_YARN_TAGSLLAMASLAtrueproject_namegalaxi}
EXECUTOR_MEMORY_OVERHEAD{EXECUTOR_MEMORY_OVERHEAD8192}
DRIVER_MEMORY_OVERHEAD{DRIVER_MEMORY_OVERHEAD1024}
BROADCAST_JOIN_THRESHOLD{BROADCAST_JOIN_THRESHOLD104857600}
SHUFFLE_PARTITIONS{SHUFFLE_PARTITIONS6001}
SPARK_DYNAMICALLOCATION_ENABLED{SPARK_DYNAMICALLOCATION_ENABLEDtrue}
SPARK_DYNAMICALLOCATION_MINEXECUTORS{SPARK_DYNAMICALLOCATION_MINEXECUTORS10}
SPARK_DYNAMICALLOCATION_MAXEXECUTORS{SPARK_DYNAMICALLOCATION_MAXEXECUTORS500}
NUM_EXECUTORS{NUM_EXECUTORS{SPARK_DYNAMICALLOCATION_MINEXECUTORS}}
SPARK_MEMORY_FRACTION{SPARK_MEMORY_FRACTION06}
SPARK_SHUFFLE_SORT_BYPASSMERGETHRESHOLD{SPARK_SHUFFLE_SORT_BYPASSMERGETHRESHOLD200}
SQL_OUTPUT_PARTITIONS{SQL_OUTPUT_PARTITIONS200}
SPARK_SHUFFLE_SERVICE_ENABLED{SPARK_DYNAMICALLOCATION_ENABLED}

umask 000

input

while [ 1 ] do
if [ 1 i ] then
shift
# readlinklinux找出符号链接指位置
file(readlink f 1)
export full_name(dirname {file})
export files`find {full_name} regex *＼＼(py) | paste sd `
PY_NAME`basename 1`
input1
fi
shift
done

PY_NAME`echo 1 | grep o ＇^[^＼]*＇`
CUR_PATH(cd (dirname 0)pwd)
echo CUR_PATH

DEPLOY_MODE{{cluster|client}}
JOB_NAMEPY{PY_NAME}{USER}_`date +s`

{SPARK_HOME}binsparksubmit ＼
master yarn ＼
deploymode cluster ＼
name {JOB_NAME} ＼
queue {QUEUE} ＼
executormemory {SPARK_EXECUTOR_MEMORY} ＼
executorcores {SPARK_EXECUTOR_CORE} ＼
conf sparkdriverextraJavaOptionsDhdpversion{HADOOP_VERSION} Dhadoop{HADOOP_CONF_DIR} Dlog4jconfigurationlog4jproperties DLOG_DIR{LOG_DIR} DJOB_NAME{JOB_NAME} DEMAIL_SERVER{EMAIL_SERVER} DEMAIL_FROM{EMAIL_FROM} DEMAIL_TO{EMAIL_TO} ＼
conf sparkexecutorextraJavaOptionsDhdpversion{HADOOP_VERSION} XX+PrintGCDateStamps XX+PrintFlagsFinal XX+PrintGCDetails XX+PrintGC XX+PrintGCTimeStamps ＼
conf sparkyarnamextraJavaOptionsDhdpversion{HADOOP_VERSION} ＼
conf sparksqlautoBroadcastJoinThreshold{BROADCAST_JOIN_THRESHOLD} ＼
conf sparkdrivermemory{SPARK_DRIVER_MEMORY} ＼
conf sparkdrivercores{SPARK_DRIVER_CORE} ＼
conf sparkdriverextraClassPathusrhdpcurrenthadoopclientlibsnappy*jar ＼
conf sparkdriverextraLibraryPathusrhdpcurrenthadoopclientlibnative ＼
conf sparkexecutorextraLibraryPathusrhdpcurrenthadoopclientlibnative ＼
conf sparkyarndrivermemoryOverhead{DRIVER_MEMORY_OVERHEAD} ＼
conf sparkyarnexecutormemoryOverhead{EXECUTOR_MEMORY_OVERHEAD} ＼
conf sparkyarnmaxAppAttempts1 ＼
conf sparkshuffleiopreferDirectBufsfalse ＼
conf sparkdrivermaxResultSize{SPARK_DRIVER_MEMORY} ＼
conf sparktaskmaxFailures10 ＼
conf sparknetworktimeout600s ＼
conf sparksqlshufflepartitions{SHUFFLE_PARTITIONS} ＼
conf sparkyarnstagingDir{STAGE_DIR} ＼
conf sparkhadoopyarntimelineserviceenabledfalse ＼
conf sparkdynamicAllocationenabled{SPARK_DYNAMICALLOCATION_ENABLED} ＼
conf sparkdynamicAllocationminExecutors{SPARK_DYNAMICALLOCATION_MINEXECUTORS} ＼
conf sparkdynamicAllocationmaxExecutors{SPARK_DYNAMICALLOCATION_MAXEXECUTORS} ＼
conf sparkdynamicAllocationexecutorIdleTimeout3600s ＼
conf sparkdynamicAllocationschedulerBacklogTimeout600s ＼
conf sparkyarntagsLLAMASLAtrueproject_name<项目名称> ＼
numexecutors {NUM_EXECUTORS} ＼
conf sparkmemoryfraction{SPARK_MEMORY_FRACTION} ＼
conf sparkshufflesortbypassMergeThreshold{SPARK_SHUFFLE_SORT_BYPASSMERGETHRESHOLD} ＼
conf sqloutputpartitions{SQL_OUTPUT_PARTITIONS} ＼
conf sparkshuffleserviceenabled{SPARK_SHUFFLE_SERVICE_ENABLED} ＼
conf sparkyarnaccesshadoopFileSystemshdfs ＼
files {files}{HADOOP_CONF_DIR}hdfssitexml{SPARK_CONF_DIR}hivesitexml{CONF_DIR}zookeeperproperties{CONF_DIR}dragonkeytab{CONF_DIR}graphjson{CONF_DIR}tablejson ＼
[principal <账号>@<密钥分发中心 KDC> ＼
keytab <存储路径><认证文件>keytab ＼]
archives <存储路径><包Python虚拟环境>zip ＼
conf sparkyarnappMasterEnvPYSPARK_PYTHONusrbinpython3 ＼
input

运行Shell：
sh [<存储路径>]sh i py

Shell脚容：
vim [<存储路径>]sh

for i in @
do
case i in
i*|input*)
INPUT{i#*}
shift # past argumentvalue

*)

esac
done

echo input sql path{INPUT}

commandsh [<存储路径>]sh i {INPUT}
echo {command}
eval command 2
res
if [ {res}X 0X ]then
echo INFO Run PySpark file successfully
else
echo ERROR Run PySpark failed and check log for detail please
fi
return res

运行Shell：
sh [<存储路径>]sh ipy

readlink：

readlinklinux找出符号链接指位置
例1：
readlink f usrbinawk
结果：
usrbingawk #usrbinawk软连接指gawk
例2：
readlink f homesoftwarelog
homesoftwarelog #果没链接显示身绝路径

获取前脚路径：
pathsh
#binbash
path(cd `dirname 0`pwd)
echo path
path2(dirname 0)
echo path2
前脚存路径：homesoftware
sh pathsh
homesoftware

解释：
dirname 0 获取前脚相路径
cd `dirname 0`pwd 先cd前路径然pwd印成绝路径

方法二：
pathsh
#binbash
path(dirname 0)
path2(readlink f path)
echo path2
sh pathsh
homesoftware
解释：
readlink f path 果path没链接显示身绝路径
获取路径较
pathsh
#binbash
PATH1(dirname 0)
PATH2(cd `dirname 0`pwd)
PATH3(readlink f PATH1)
echo PATH1
echo PATH2
echo PATH3
前脚存路径：homesoftware
sh pathsh
echo PATH1
homesoftware echo PATH2
home echo PATH3
Shell命令行交互式运行代码
mysql [h ] [u <账号>] [p<密码>] [<数库名>]
mysql><查询语句数操作语句1>

mysql><查询语句数操作语句2>

mysql><查询语句数操作语句3>

…

mysql><查询语句数操作语句n>
mysql>QUIT
usrbinbteq c UTF8
LOGON <账号><密码>

<查询语句数操作语句1>

<查询语句数操作语句2>

<查询语句数操作语句3>

…

<查询语句数操作语句n>

IF ERRORCODE <> 0 THEN QUIT ERRORCODE

LOGOFF
QUIT

BTEQBasic Teradata QueryTeradata发行提交SQL查询前端工具BTEQ命令必须开头结尾者什

BTEQ常报表格式化输出设置：
SET DEFAULTS：输出格式定义成默认值
SET ECHOREQ ONOFF：否SQL请求BTEQ命令复制输出报表中
SET FOLDLINE ON 1：第1字段显示第1行字段值显示第2行
SET FOOTING [NULL＇string＇]：定义脚注包含&DATE&TIME&PAGE&n
SET FORMAT ONOFF：设置OFF时BTEQ忽略FOOTINGFORMCHARRTITLEHEADINGPAGEBREAK等设置
SET HEADING [NULL＇string＇]：定义页头FOOTING样
SET NULL AS ＇string＇：NULL默认值问号改变缺省值
SET OMIT ONOFF [nALL]：指定字段包括报表中页头脚注中
SET PAGEBREAK ONOFF [nALL]：指定字段值发生变化时插入分页符开始新页
SET PAGELENGTH n：定义页面长度默认值55行
SET RTITLE [＇string＇]：定义页方标题动包含日期页号
SET SEPARATOR [＇string＇n]：定义字段间分隔符n表示空格
SET SUPPRESS ONOFF [nALL]：指定字段果遇连续重复值空格代
SET SKIPLINE ONOFF [nALL]：指定字段值发生变化时插入空行
SET SKIPDOUBLE ONOFF [nALL]：指定字段值发生变化时插入两空行
SET UNDERLINE ONOFF [nALL]：行输出中指定字段加划线
SET WIDTH n：设置报表宽度默认值75

假定Teradata数库DEMO(名字必须HOSTS文件中进行定义)SQL01户名进行登录键入logon demosql01退出Teradata命令logoff退出BTEQ命令quit果需BTEQ中运行unix命令必须运行os xxxx

BTEQ交互方式运行批处理方式运行BTEQ输出保存文件中重新恢复标准输出：export filexxxxexport reset

编写BTEQ脚时插入行进行注释采单行进行注释：
SET SESSION TRANSACTION ANSI
脚文件中必须户密码logon命令中提供
LOGIN sql01sql01
SELECT FROM WHERE
QUIT
保存脚testScript通run filetestScript进行脚运行
binpyspark
>>>

>>>

>>>

…

>>>
>>> exit()

hive
hive>

hive>

hive>

…
hive>

hive> exit

导入导出CSV文件
导入：
LOAD DATA INFILE ＇<存储路径>＼<文件名>csv＇ INTO TABLE [`<库名>`] `<表名>` FIELDS TERMINATED BY ＇＇ OPTIONALLY ENCLOSED BY ＇＇ LINES TERMINATED BY ＇＼n＇
常参数：
FIELDS TERMINATED BY ＇＇：指定字段分隔符
OPTIONALLY ENCLOSED BY ＇＇：认双引号中独立字段Excel 转 CSV 时特殊字符（逗号顿号等）字段会动双引号引起
LINES TERMINATED BY ＇＼n＇：指定行分隔符注意 Windows 台创建文件分隔符＇＼r＼n＇

导出：
mysql > SELECT <字段名1><字段名2><字段名3><字段名n> INTO OUTFILE ＇<存储路径>＼<文件名>csv＇
FIELDS TERMINATED BY ＇＇ OPTIONALLY ENCLOSED BY ＇＇
LINES TERMINATED BY ＇＼n＇
FROM [`<库名>`] `<表名>`
[筛选条件]
导入：
编辑文件：
vim <文件名>in
SET width 64000
SET session transaction btet
logmech ldap
logon <户名><密码>

DATABASE <库名>

PACK 1000
IMPORT VARTEXT ＇＇ FILE<存储路径>＼<文件名>csv
REPEAT *
USING(<字段名1> <类型1>
<字段名2> <类型2>
<字段名3> <类型3>
…
<字段名n> <类型n>)

insert into <库名><表名> (
<字段名1>
<字段名2>
<字段名3>

<字段名n>
)
values
( <字段名1>
<字段名2>
<字段名3>

<字段名n>
)
LOGOFF
EXIT

执行文件：
binbteq < <文件名>in

导出：
vim <文件名>out
SET SESSION TRANSACTION BTET
LOGON <户名><密码>
EXPORT FILE <存储路径>＼<文件名>csv
SET SEPARATOR ＇＇
DATABASE <库名>
SELECT * FROM <表名>
[筛选条件]

LOGOFF
EXIT

执行文件：
binbteq < <文件名>out
导入：
df sparkreadload(
path＇＇
format＇csv＇ headerTrue)

导出：
df…
dfrepartition(1)writecsv(path＇<存储路径><表名>[<分区字段><分区值>]csv＇ headerTrue sep mode＇overwrite＇)

dfwriteformat(comdatabrickssparkcsv’)save(<存储路径><表名>[<分区字段><分区值>]csv’)

dftoPandas()to_csv(<存储路径><表名>[<分区字段><分区值>]csv’)

index否索引header否列名True需
outputpath<存储路径><表名>[<分区字段><分区值>]csv’
dfto_csv(outputpathsep’’indexFalseheaderFalse)

#方法

df sparkreadcsv(r<存储路径><表名>[<分区字段><分区值>]csv encoding＇gbk＇ headerTrue inferSchemaTrue) # header表示数第行否列名inferSchema表示动推断schema时未指定schema

者：

df sparkreadcsv(r<存储路径><表名>[<分区字段><分区值>]csv encoding＇gbk＇ headerTrue schemaschema)#指定schema

#方法二

df sparkreadformat(csv)option(headerTrue)option(encoding gbk)load(r<存储路径><表名>[<分区字段><分区值>]csv)

者：

df sparkreadformat(csv)option(encodinggbk)option(headerTrue)load(r<存储路径><表名>[<分区字段><分区值>]csv schemaschema)

# 写csv例追加数格式：

dfwritemode(＇append＇)option()option()format()save()

#注意：数建立csv第行列名情况列名时应该掉header属性
权限控制
查询户权限：
Global level privileges
SELECT CONCAT(user ＇@＇ host)delete_privdrop_priv FROM mysqluser

Table level privileges
select CONCAT(user ＇@＇ host)usertable from mysqltables_priv

SHOW GRANTS
查询户权限：
SELECT
UserName
DatabaseName
TableName
ColumnName
AccessRight
GrantAuthority
GrantorName
AllnessFlag
CreatorName
CreateTimeStamp
FROM dbcallrights
WHERE username＇<户ID>＇
AND databasename＇<库名>＇

查询户权限
execute <库名>AllUserRights (＇<户名>＇)
UDF宏定义
create macro <库名>AllUserRights (UserName char(128)) as (
locking row for access select
UserName (varchar(128))
AccessType (varchar(128))
RoleName (varchar(128))
DatabaseName (varchar(128))
TableName (varchar(128))
ColumnName (varchar(128))
AccessRight
case
when accessright＇AE＇ then ＇ALTER EXTERNALPROCEDURE＇
when accessright＇AF＇ then ＇ALTER FUNCTION＇
when accessright＇AP＇ then ＇ALTER PROCEDURE＇
when accessright＇AS＇ then ＇ABORT SESSION＇
when accessright＇CA＇ then ＇CREATE AUTHORIZATION＇
when accessright＇CD＇ then ＇CREATE DATABASE＇
when accessright＇CE＇ then ＇CREATE EXTERNAL PROCEDURE＇
when accessright＇CF＇ then ＇CREATE FUNCTION＇
when accessright＇CG＇ then ＇CREATE TRIGGER＇
when accessright＇CM＇ then ＇CREATE MACRO＇
when accessright＇CO＇ then ＇CREATE PROFILE＇
when accessright＇CP＇ then ＇CHECKPOINT＇
when accessright＇CR＇ then ＇CREATE ROLE＇
when accessright＇CS＇ then ＇CREATE SERVER＇
when accessright＇CT＇ then ＇CREATE TABLE＇
when accessright＇CU＇ then ＇CREATE USER＇
when accessright＇CV＇ then ＇CREATE VIEW＇
when accessright＇CZ＇ then ＇CREATE ZONE＇
when accessright＇C1＇ then ＇CREATE DATASET SCHEMA＇
when accessright＇D＇ then ＇DELETE＇
when accessright＇DA＇ then ＇DROP AUTHORIZATION＇
when accessright＇DD＇ then ＇DROP DATABASE＇
when accessright＇DF＇ then ＇DROP FUNCTION＇
when accessright＇DG＇ then ＇DROP TRIGGER＇
when accessright＇DM＇ then ＇DROP MACRO＇
when accessright＇DO＇ then ＇DROP PROFILE＇
when accessright＇DP＇ then ＇DUMP＇
when accessright＇DR＇ then ＇DROP ROLE＇
when accessright＇DS＇ then ＇DROP SERVER＇
when accessright＇DT＇ then ＇DROP TABLE＇
when accessright＇DU＇ then ＇DROP USER＇
when accessright＇DV＇ then ＇DROP VIEW＇
when accessright＇DZ＇ then ＇DROP ZONE＇
when accessright＇D1＇ then ＇DROP DATASET SCHEMA＇
when accessright＇E＇ then ＇EXECUTE＇
when accessright＇EF＇ then ＇EXECUTE FUNCTION＇
when accessright＇GC＇ then ＇CREATE GLOP＇
when accessright＇GD＇ then ＇DROP GLOP＇
when accessright＇GM＇ then ＇GLOP MEMBER＇
when accessright＇I＇ then ＇INSERT＇
when accessright＇IX＇ then ＇INDEX＇
when accessright＇MC＇ then ＇CREATE MAP＇
when accessright＇MD＇ then ＇DROP MAP＇
when accessright＇MR＇ then ＇MONITOR RESOURCE＇
when accessright＇MS＇ then ＇MONITOR SESSION＇
when accessright＇NT＇ then ＇NONTEMPORAL＇
when accessright＇OD＇ then ＇OVERRIDE DELETE POLICY＇
when accessright＇OI＇ then ＇OVERRIDE INSERT POLICY＇
when accessright＇OP＇ then ＇CREATE OWNER PROCEDURE＇
when accessright＇OS＇ then ＇OVERRIDE SELECT POLICY＇
when accessright＇OU＇ then ＇OVERRIDE UPDATE POLICY＇
when accessright＇PC＇ then ＇CREATE PROCEDURE＇
when accessright＇PD＇ then ＇DROP PROCEDURE＇
when accessright＇PE＇ then ＇EXECUTE PROCEDURE＇
when accessright＇R＇ then ＇RETRIEVESELECT＇
when accessright＇RF＇ then ＇REFERENCES＇
when accessright＇RS＇ then ＇RESTORE＇
when accessright＇SA＇ then ＇SECURITY CONSTRAINT ASSIGNMENT＇
when accessright＇SD＇ then ＇SECURITY CONSTRAINT DEFINITION＇
when accessright＇ST＇ then ＇STATISTICS＇
when accessright＇SS＇ then ＇SET SESSION RATE＇
when accessright＇SR＇ then ＇SET RESOURCE RATE＇
when accessright＇TH＇ then ＇CTCONTROL＇
when accessright＇U＇ then ＇UPDATE＇
when accessright＇UU＇ then ＇UDT Usage＇
when accessright＇UT＇ then ＇UDT Type＇
when accessright＇UM＇ then ＇UDT Method＇
when accessright＇W1＇ then ＇WITH DATASET SCHEMA＇
when accessright＇ZO＇ then ＇ZONE OVERRIDE＇
else＇＇
end (varchar(26)) as AccessRightDesc
GrantAuthority
GrantorName (varchar(128))
AllnessFlag
CreatorName (varchar(128))
CreateTimeStamp
from
(
select
UserName
＇User＇ (varchar(128)) as AccessType
＇＇ (varchar(128)) as RoleName
DatabaseName
TableName
ColumnName
AccessRight
GrantAuthority
GrantorName
AllnessFlag
CreatorName
CreateTimeStamp
from dbcallrights
where UserName username
and CreatorName not username
union all
select
Grantee as UserName
＇Member＇ as UR
rRoleName
DatabaseName
TableName
ColumnName
AccessRight
null (char(1)) as GrantAuthority
GrantorName
null (char(1)) as AllnessFlag
null (char(1)) as CreatorName
CreateTimeStamp
from dbcallrolerights r
join dbcrolemembers m
on mRoleName rRoleName
where UserName username
union all
select
User as UserName
mGrantee as UR
rRoleName
DatabaseName
TableName
ColumnName
AccessRight
null (char(1)) as GrantAuthority
GrantorName
null (char(1)) as AllnessFlag
null (char(1)) as CreatorName
CreateTimeStamp
from dbcallrolerights r
join dbcrolemembers m
on mRoleName rRoleName
where mgrantee in (select rolename from dbcrolemembers where grantee
username)
) AllRights
order by 4567 )

通dbcallrights表中UserName列DatabaseName列TableName列AccessRight列查询获取指定指定数库中指定表操作权限执行某条SQL语句前判定前户否执行语句权限权限足时尝试动授权（太安全执行完应revoke）等措施
AccessRight列缩写词应列表（40）：

AccessRight
含义
AF
ALTER FUNCTION
AP
ALTER PROCEDURE
AS
ABORT SESSION
CD
CREATE DATABASE
CF
CREATE FUNCTION
CG
CREATE TRIGGER
CM
CREATE MACRO
CO
CREATE PROFILE
CP
CHECKPOINT
CR
CREATE ROLE
CT
CREATE TABLE
CU
CREATE USER
CV
CREATE VIEW
D
DELETE
DD
DROP DATABASE
DF
DROP FUNCTION
DG
DROP TRIGGER
DM
DROP MACRO
DO
DROP PROFILE
DP
DUMP
DR
DROP ROLE
DT
DROP TABLE
DU
DROP USER
DV
DROP VIEW
E
EXECUTE
EF
EXECUTE FUNCTION
I
INSERT
IX
INDEX
MR
MONITOR RESOURCE
MS
MONITOR SESSION
PC
CREATE PROCEDURE
PD
DROP PROCEDURE
PE
EXECUTE PROCEDURE
RO
REPLICATION OVERRIDE
R
RETRIEVESELECT
RF
REFERENCE
RS
RESTORE
SS
SET SESSION RATE
SR
SET RESOURCE RATE
U
UPDATE
示例SQL语句：

select username databasename tablename accessright from dbcallrights
where databasename＇systemfe＇ and username＇dbc＇ and tablename＇opt_ras_table＇
述语句执行结果：
*** Query completed 12 rows found 4 columns returned
*** Total elapsed time was 1 second

UserName DatabaseName TableName AccessRight

DBC SystemFe opt_ras_table DT
DBC SystemFe opt_ras_table U
DBC SystemFe opt_ras_table DG
DBC SystemFe opt_ras_table RF
DBC SystemFe opt_ras_table RS
DBC SystemFe opt_ras_table R
DBC SystemFe opt_ras_table I
DBC SystemFe opt_ras_table CG
DBC SystemFe opt_ras_table ST
DBC SystemFe opt_ras_table DP
DBC SystemFe opt_ras_table D
DBC SystemFe opt_ras_table IX
SQL语句动构建出授予权限SQL语句（ GRANT语句）：

SEL
TRIM(username)
TRIM(databasename)
TRIM(tablename)
＇GRANT ＇|| CASE
WHEN AccessRight ＇AF ＇ THEN ＇ALTER FUNCTION＇
WHEN AccessRight ＇AP ＇ THEN ＇ALTER PROCEDURE＇
WHEN AccessRight ＇AS ＇ THEN ＇ABORT SESSION＇
WHEN AccessRight ＇CD ＇ THEN ＇CREATE DATABASE＇
WHEN AccessRight ＇CF ＇ THEN ＇CREATE FUNCTION＇
WHEN AccessRight ＇CG ＇ THEN ＇CREATE TRIGGER＇
WHEN AccessRight ＇CM ＇ THEN ＇CREATE MACRO＇
WHEN AccessRight ＇CO ＇ THEN ＇CREATE PROFILE＇
WHEN AccessRight ＇CP ＇ THEN ＇CHECKPOINT＇
WHEN AccessRight ＇CR ＇ THEN ＇CREATE ROLE＇
WHEN AccessRight ＇CT ＇ THEN ＇CREATE TABLE＇
WHEN AccessRight ＇CU ＇ THEN ＇CREATE USER＇
WHEN AccessRight ＇CV ＇ THEN ＇CREATE VIEW＇
WHEN AccessRight ＇D ＇ THEN ＇DELETE＇
WHEN AccessRight ＇DD ＇ THEN ＇DROP DATABASE＇
WHEN AccessRight ＇DF ＇ THEN ＇DROP FUNCTION＇
WHEN AccessRight ＇DG ＇ THEN ＇DROP TRIGGER＇
WHEN AccessRight ＇DM ＇ THEN ＇DROP MACRO＇
WHEN AccessRight ＇DO ＇ THEN ＇DROP PROFILE＇
WHEN AccessRight ＇DP ＇ THEN ＇DUMP＇
WHEN AccessRight ＇DR ＇ THEN ＇DROP ROLE＇
WHEN AccessRight ＇DT ＇ THEN ＇DROP TABLE＇
WHEN AccessRight ＇DU ＇ THEN ＇DROP USER＇
WHEN AccessRight ＇DV ＇ THEN ＇DROP VIEW＇
WHEN AccessRight ＇E ＇ THEN ＇EXECUTE＇
WHEN AccessRight ＇EF ＇ THEN ＇EXECUTE FUNCTION＇
WHEN AccessRight ＇I ＇ THEN ＇INSERT＇
WHEN AccessRight ＇IX ＇ THEN ＇INDEX＇
WHEN AccessRight ＇MR ＇ THEN ＇MONITOR RESOURCE＇
WHEN AccessRight ＇MS ＇ THEN ＇MONITOR SESSION＇
WHEN AccessRight ＇PC ＇ THEN ＇CREATE PROCEDURE＇
WHEN AccessRight ＇PD ＇ THEN ＇DROP PROCEDURE＇
WHEN AccessRight ＇PE ＇ THEN ＇EXECUTE PROCEDURE＇
WHEN AccessRight ＇RO ＇ THEN ＇REPLICATION OVERRIDE＇
WHEN AccessRight ＇R ＇ THEN ＇RETRIEVESELECT＇
WHEN AccessRight ＇RF ＇ THEN ＇REFERENCE＇
WHEN AccessRight ＇RS ＇ THEN ＇RESTORE＇
WHEN AccessRight ＇SS ＇ THEN ＇SET SESSION RATE＇
WHEN AccessRight ＇SR ＇ THEN ＇SET RESOURCE RATE＇
WHEN AccessRight ＇U ＇ THEN ＇UPDATE＇
END || ＇ ON ＇||TRIM(databasename)||＇＇||TRIM(tablename)||＇ to ＇||TRIM(username)||＇＇ AS Permission
FROM dbcAllRights
WHERE DatabaseName ＇<库名>＇ and USERNAME ＇<户名>＇ AND TABLENAME ＇<表名>＇
Hive表中：
认证(authentication)：验证户身份否
授权(authorization)：验证户身份操作否权限
目前hive（版0120）支持简单权限理默认情况开启样户具相权限时超级理员hive中表查改动权利样符合般数仓库安全原Hive基元数权限理基文件存储级权限理次介绍MetaData权限理通配置开启Hive身份认证功进行权限检查：

配置
1开启启身份认证户必须grant privilege实体进行操作
hivesecurityauthorizationenabled true

2表示创建表时动赋予户角色相应权限
hivesecurityauthorizationcreatetableownergrants ALL
hivesecurityauthorizationcreatetablerolegrants admin_roleALL
hivesecurityauthorizationcreatetableusergrants user1user2selectuser3create

3< 假出现错误： Error while compiling statement FAILED SemanticException The current builtin authorization in Hive is incomplete and disabled 需配置面属性 >
hivesecurityauthorizationtaskfactory orgapachehadoophiveqlparseauthorizationHiveAuthorizationTaskFactoryImpl

角色理
创建删角色
create role role_name
drop role role_name
展示roles
show roles
赋予角色权限
grant select on database db_name to role role_name
grant select on [table] t_name to role role_name
查角色权限
show grant role role_name on database db_name
show grant role role_name on [table] t_name
角色赋予户
grant role role_name to user user_name
回收角色权限
revoke select on database db_name from role role_name
revoke select on [table] t_name from role role_name
查某户角色
show role grant user user_name

超级权限
Hive权限功需完善方超级理员
Hive中没超级理员户进行GrantRevoke操作完善超级理员必须添加hivesemanticanalyzerhook配置实现权限控制类

hivesemanticanalyzerhook commycompanyAuthHook

编译面代码(需导入赖antlrruntime34jarhiveexec0120cdh512jar)
包成jar放置hiveclasspath(客户端hive shell机hiveenvsh 中环境变量：HIVE_AUX_JARS_PATH指路径配置仅hive shell生效)
hivesitexml中添加参数hiveauxjarspath(目前仅支持路径) fileusrlibhivelibHiveAuthHookjar(配置仅hive server效)重启hiveserver

package comnewland

import orgapachehadoophiveqlparseASTNode
import orgapachehadoophiveqlparseAbstractSemanticAnalyzerHook
import orgapachehadoophiveqlparseHiveParser
import orgapachehadoophiveqlparseHiveSemanticAnalyzerHookContext
import orgapachehadoophiveqlparseSemanticException
import orgapachehadoophiveqlsessionSessionState

public class AuthHook extends AbstractSemanticAnalyzerHook {
private static String[] admin { root hadoop }

@Override
public ASTNode preAnalyze(HiveSemanticAnalyzerHookContext context
ASTNode ast) throws SemanticException {
switch (astgetToken()getType()) {
case HiveParserTOK_CREATEDATABASE
case HiveParserTOK_DROPDATABASE
case HiveParserTOK_CREATEROLE
case HiveParserTOK_DROPROLE
case HiveParserTOK_GRANT
case HiveParserTOK_REVOKE
case HiveParserTOK_GRANT_ROLE
case HiveParserTOK_REVOKE_ROLE
String userName null
if (SessionStateget() null
&& SessionStateget()getAuthenticator() null) {
userName SessionStateget()getAuthenticator()getUserName()
}
if (admin[0]equalsIgnoreCase(userName)
&& admin[1]equalsIgnoreCase(userName)) {
throw new SemanticException(userName
+ can＇t use ADMIN options except + admin[0] +
+ admin[1] + )
}
break
default
break
}
return ast
}

public static void main(String[] args) throws SemanticException {
String[] admin { admin root }
String userName root
for (String tmp admin) {
Systemoutprintln(tmp)
if (tmpequalsIgnoreCase(userName)) {
throw new SemanticException(userName
+ can＇t use ADMIN options except + admin[0] +
+ admin[1] + )
}
}
}
}

HIVE支持权限：

权限名称含义
ALL 权限
ALTER 允许修改元数（modify metadata data of object）表信息数
UPDATE 允许修改物理数（modify physical data of object）实际数
CREATE 允许进行Create操作
DROP 允许进行DROP操作
INDEX 允许建索引（目前没实现）
LOCK 出现发允许户进行LOCKUNLOCK操作
SELECT 允许户进行SELECT操作
SHOW_DATABASE 允许户查数库
附：
登录hive元数库发现表
Db_privs记录UserRoleDB权限
Tbl_privs记录UserRoletable权限
Tbl_col_privs：记录UserRoletable column权限
Roles：记录创建role
Role_map：记录UserRole应关系
行列转换
列转行：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
SUM(CASE <分类字段> WHEN <分类值1> THEN <度量字段> ELSE 0 END) AS <分类值名1>
SUM(CASE <分类字段> WHEN <分类值2> THEN <度量字段> ELSE 0 END) AS <分类值名2>
SUM(CASE <分类字段> WHEN <分类值3> THEN <度量字段> ELSE 0 END) AS <分类值名3>
…
SUM(CASE <分类字段> WHEN <分类值n> THEN <度量字段> ELSE 0 END) AS <分类值名n>
FROM [`<架构名称>`] `<表名>`
GROUP BY <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>

SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<分类字段><度量字段1> <度量字段2> <度量字段3>… <度量字段n>
FROM [`<架构名称>`] `<表名>`
PIVOT(SUM(<度量字段1>) AS <度量字段1>SUM(<度量字段2>) AS <度量字段2>SUM(<度量字段3>) AS <度量字段3>…SUM(<度量字段n>) AS <度量字段n> FOR <分类字段> IN (<分类值1><分类值2><分类值3>…<分类值n>))

SET @sql NULL
SELECT
GROUP_CONCAT(DISTINCT
CONCAT(
＇SUM(CASE <分类字段> WHEN ＇＇＇
<分类字段>
＇＇＇ THEN IFNULL(<度量字段>0) ELSE 0 END) AS `＇
<分类字段> ＇`＇
)
) INTO @sql
FROM
(
SELECT DISTINCT <分类字段>
FROM [`<架构名称>`] `<表名>`
ORDER BY <分类字段>
) T

SET @sql
CONCAT(＇SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
＇
@sql
＇ FROM [`<架构名称>`] `<表名>`
GROUP BY <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>＇)

PREPARE stmt FROM @sql
EXECUTE stmt
DEALLOCATE PREPARE stmt

列转行分类值字符串连接：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
GROUP_CONCAT(TRIM(<分类字段>)) AS <分类字段> GROUP_CONCAT(CAST(<度量字段> AS VARCHAR(30))) AS <度量字段>
FROM [`<架构名称>`] `<表名>`
GROUP BY <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>

行转列：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<分类字段><度量字段> from [`<架构名称>`] `<表名>`
UNPIVOT
(<度量字段> FOR <分类字段> IN (<分类值1><分类值2><分类值3>…<分类值n>))

逗号分隔数拆分成行：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
substring_index(substring_index(a<带逗号数字段>＇＇bhelp_topic_id+1)＇＇1)
FROM [`<架构名称>`] `<表名>` a
JOIN mysqlhelp_topic b
ON bhelp_topic_id < (length(a<带逗号数字段>) length(replace(a<带逗号数字段>＇＇＇＇))+1)
ORDER BY <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
列转行：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<分类字段><度量字段1> <度量字段2> <度量字段3>… <度量字段n>
FROM [<架构名称>] <表名>
PIVOT(SUM(<度量字段1>) AS <度量字段1>SUM(<度量字段2>) AS <度量字段2>SUM(<度量字段3>) AS <度量字段3>…SUM(<度量字段n>) AS <度量字段n> for <分类字段> IN (<分类值1><分类值2><分类值3>…<分类值n>))

列转行分类值字符串连接：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
CAST(tdstatsudfconcat(TRIM(<分类字段>)) AS varchar(500)) AS <分类字段>tdstatsudfconcat(CAST(<度量字段> AS VARCHAR(500))) AS <度量字段>
FROM [<架构名称>] <表名>
GROUP BY 123…n

SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
TRIM(TRAILING ＇＇ FROM (XMLAGG(<分类字段> || ＇＇)(VARCHAR(500)))) AS <分类字段>TRAILING ＇＇ FROM (XMLAGG(CAST(<度量字段> AS VARCHAR(500)) || ＇＇)) AS <度量字段>
FROM [<架构名称>] <表名>
GROUP BY 123…n

行转列：
SELECT <维度字段1>
<维度字段2>
<维度字段3>
…
<维度字段n>
<分类字段><度量字段1> <度量字段2> <度量字段3>… <度量字段n>
FROM [<架构名称>] <表名> UNPIVOT [{INCLUDE|EXCLUDE} NULLS] (
(<度量字段1> <度量字段2> <度量字段3>… <度量字段n>)
FOR <分类字段> IN (
(<分类值1> <分类值2> <分类值3>…<分类值n>) AS ＇<分类值123…n名>＇
(<分类值n+1> <分类值n+2> <分类值n+3>…<分类值2n>) AS ＇<分类值n+1n+2n+3…2n名>＇
(<分类值2n+1> <分类值2n+2> <分类值2n+3>…<分类值3n>) AS ＇<分类值2n+12n+22n+3…3n名>＇
…
(<分类值mn+1> <分类值mn+2> <分类值mn+3>…<分类值(m+1)n>) AS ＇<分类值mn+1mn+2mn+3…(m+1)n名>＇
)
) T

逗号分隔数拆分成行：
USE [<架构名称>]

SELECT A* FROM TABLE (strtok_split_to_table( <表名><维度字段1>
<表名><维度字段2>
<表名><维度字段3>
…
<表名><维度字段n>
<表名><带逗号数字段> ＇＇)
RETURNS (<维度字段名1> <维度字段类型1>
<维度字段名2> <维度字段类型2>
<维度字段名3> <维度字段类型3>
…
<维度字段名n> <维度字段类型n>
<带逗号数字段名>_num integer <带逗号数字段名> varchar(100) character set unicode) ) AS A
ORDER BY 123…n
PySpark里：
列转行：
import pysparksqlfunctions as func

<表名>_df…

企业大数据基础平台搭建和实用开发代码

相关文档

AE开发实例代码总结

IOS开发环境搭建

数据结构大作业（含源代码）

基于Action的数据分析大数据平台

实用版本技术开发合同

关于异地开发中的源代码管理问题

敏捷开发中高质量Java代码开发实践

搭建网络问政平台

搭建一个管理平台

领导之前要先搭建信任平台

国家队信息化平台数据库软件设计与开发合同

数据交换平台软件产品开发项目可行性报告（v3）

国家队信息化平台数据库软件设计与开发合同

授信基础数据

公平与爱是构建和谐班级的基础（论文）

开发云平台提升企业信息应用和管理水平

锐捷教你轻松搭建经济实用网络平台-交换机解决方案

搭建全员素质提升平台构筑人才强企战略高地

搭建平台鼓励非公企业积极承担社会责任

国家社科基金项目申报数据代码表（2022年）

文档贡献者

该用户的其他文档

相关PPT

相关PDF