zakaban的使用

azkaban任务文件的注意事项：

• 文件后缀需要是job，文件格式为无bom的utf8；
• 写完任务文件后需要压缩为zip文件上传到azkaban的web界面以执行；
• 任务文件中任何相对路径的引用文件需要和任务文件一起压缩到zip中。

简单任务

定义一个单任务，名称为foo，内容如下：

type=command
command=echo 'foo'

任务流

任务流就是任务之间存在依赖关系（通过dependencies来定义依赖关系），需要把有依赖关系的任务文件压缩到同一个zip中，如下：
第一个job：foo.job

type=command
command=echo 'foo'

第二个job：bar.job依赖foo.job

type=command
dependencies=foo
command=echo 'bar'

通过azkaban操作hdfs

创建job描述文件fs.job

type=command
command=/export/servers/hadoop-2.6.0-cdh5.14.0/bin/hadoop fs -mkdir /azkaban

通过zakaban操作mapreduce

创建job描述文件

type=command
command=/export/servers/hadoop-2.6.0-cdh5.14.0/bin/hadoop jar hadoop-mapreduce-examples-2.6.0-cdh5.14.0.jar pi 3 5

注意，需要在hadoop的安装目录下的share/hadoop/mapreduce路径下找到hadoop-mapreduce-examples-2.6.0-cdh5.14.0.jar这个jar包，并将它和job文件一起压缩

通过azkaban操作hive

1、创建hive脚本，命名为hive.sql

create database if not exists azhive;
use azhive;
create table if not exists aztest(id string,name string) row format delimited fields terminated by '\t';

2、创建job文件，命名为hive.job

type=command
# 通过hive -f来执行hive.sql
command=/export/servers/hive-1.1.0-cdh5.14.0/bin/hive -f 'hive.sql'

最后，将hive.sql和hive.job一起压缩并上传到azkaban

azkaban定时任务

需要在azkaban的web界面选中任务并进入任务，点击“Execute Flow”按钮，然后点击“Schedule”按钮进入定时任务的设置界面，定时规则和Linux的crontab一样，可以使用在线Crontab生成器辅助设置。

sqoop的安装

1、解压安装包到安装目录
2、修改配置文件
①cd到安装目录下的conf目录
②cp sqoop-env-template.sh sqoop-env.sh
③修改sqoop-env.sh，修改为如下，其中HADOOP_COMMON_HOME、HADOOP_MAPRED_HOME是hadoop的安装目录，HIVE_HOME是hive的安装目录，HBASE_HOME暂时不定义。

3、添加jar包
需要的jar包是java-json.jar和mysql-connector-java-5.1.40.jar，可以在网上找，将jar包添加到sqoop安装目录下的lib下

sqoop导入操作（从关系型数据库向hdfs或hive导入）

注意，关系型数据库以MySQL为例，需要先开启外部MySQL的远程root授权

sqoop连接数据库

# 连接数据库并打印所有的database
bin/sqoop list-databases --connect jdbc:mysql://172.16.43.67:3306/ --username root --password admin
# 连接数据库并打印指定database下的所有table
bin/sqoop list-tables --connect jdbc:mysql://172.16.43.67:3306/userdb --username root --password admin

导入数据库表数据到HDFS

1、不指定目录导入

bin/sqoop import --connect jdbc:mysql://192.168.1.101:3306/userdb \
--password admin --username root \
--table emp --m 1 # --m 1 是指定本次sqoop操作使用的mapTask的个数为1

2、指定目录导入

bin/sqoop import  --connect jdbc:mysql://172.16.43.67:3306/userdb \
--username root --password admin \
--delete-target-dir --table emp  --target-dir /sqoop/emp --m 1

3、指定导入文件的分隔符和目录导入

bin/sqoop import  --connect jdbc:mysql://172.16.43.67:3306/userdb \
--username root --password admin \
--delete-target-dir --table emp  --target-dir /sqoop/emp2 --m 1 \
--fields-terminated-by '\t'

导入关系表到HIVE

1、导入到已存在的hive表中
①创建hive表

hive (default)> create database sqooptohive;
hive (default)> use sqooptohive;
hive (sqooptohive)> create external table emp_hive
(id int,name string,deg string,salary int ,dept string) 
row format delimited fields terminated by '\001';

②导入数据

bin/sqoop import --connect jdbc:mysql://172.16.43.67:3306/userdb \
--username root --password admin \
--table emp --fields-terminated-by '\001' \
--hive-import --hive-table sqooptohive.emp_hive --hive-overwrite \
--delete-target-dir --m 1

2、导入数据并自动创建hive表

# 自动创建的hive表的表名和mysql中的一致
bin/sqoop import --connect jdbc:mysql://172.16.43.67:3306/userdb \
--username root --password admin \
--table emp_conn --hive-import -m 1 --hive-database sqooptohive;

导入表数据子集

bin/sqoop import \
--connect jdbc:mysql://172.16.43.67:3306/userdb \
--username root --password admin --table emp_add \
--target-dir /sqoop/emp_add -m 1  --delete-target-dir \
--where "city = 'sec-bad'"

根据查询结果导入

注意：查询语句只能用单引号包起来，必须要有where条件，如果要导入全部数据则用where 1=1，并且查询语句必须在结尾加上and $CONDITIONS。

bin/sqoop import \
--connect jdbc:mysql://172.16.43.67:3306/userdb --username root --password admin \
--delete-target-dir -m 1 \
--query 'select phno from emp_conn where 1=1 and  $CONDITIONS' \
--target-dir /sqoop/emp_conn

增量导入

sqoop自带的增量导入

bin/sqoop import \
--connect jdbc:mysql://192.168.22.22:3306/userdb \
--username root \
--password admin \
--table emp \
--incremental append \
--check-column id \
--last-value 1202  \
-m 1 \
--target-dir /sqoop/increment

这种方式并不好用，因为数据量很大时，难以得知导入的最后一条数据是什么，所以last-value的值难以确定

实际工作中的增量导入方法

1、实际工作中创建MySQL数据表，一定要加入3个字段来追踪数据变更，分别是create_time、update_time、is_delete这3个字段，示例如下：

CREATE TABLE emp (
id INT(11) DEFAULT NULL,
name VARCHAR(100) DEFAULT NULL,
deg VARCHAR(100) DEFAULT NULL,
salary INT(11) DEFAULT NULL,
dept VARCHAR(10) DEFAULT NULL,
create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
is_delete BIGINT(20) DEFAULT '1'
) ENGINE=INNODB DEFAULT CHARSET=utf8;

2、通过查询导入增量数据

# 通过where导入
bin/sqoop import \
--connect jdbc:mysql://192.168.22.22:3306/userdb \
--username root \
--password admin  \
--table emp \
--incremental append  \
--where "create_time > '2018-06-17 00:00:00' and is_delete='1' and create_time < '2018-06-17 23:59:59'" \
--target-dir /sqoop/incement \
--check-column id  \
--m 1



# 通过query导入
bin/sqoop import \
--connect jdbc:mysql://192.168.22.22:3306/userdb \
--username root \
--password admin  \
--query 'select * from emp where create_time>2018-06-17 00:00:00 and is_delete=1 and create_time < 2018-06-17 23:59:59 and $CONDITIONS' \
--delete-target-dir --m 1 \
--target-dir /sqoop/incement

sqoop导出操作（从hdfs或hive导出到关系型数据库）

1、创建mysql表

CREATE TABLE `emp_out` (
 `id` INT(11) DEFAULT NULL,
 `name` VARCHAR(100) DEFAULT NULL,
 `deg` VARCHAR(100) DEFAULT NULL,
 `salary` INT(11) DEFAULT NULL,
 `dept` VARCHAR(10) DEFAULT NULL,
 `create_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
 `update_time` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
 `is_delete` BIGINT(20) DEFAULT '1'
) ENGINE=INNODB DEFAULT CHARSET=utf8;

2、执行导出命令

bin/sqoop export \
--connect jdbc:mysql://172.16.43.67:3306/userdb \
--username root --password admin \
--table emp_out \
--export-dir /sqoop/emp \
--input-fields-terminated-by "\t"

sqoop常用命令及参数

公用参数：数据库连接

序号	参数	说明
1	–connect	连接关系型数据库的URL
2	–connection-manager	指定要使用的连接管理类
3	–driver	JDBC的driver class
4	–help	打印帮助信息
5	–password	连接数据库的密码
6	–username	连接数据库的用户名
7	–verbose	在控制台打印出详细信息

公用参数：import

序号	参数	说明
1 -	-enclosed-by	给字段值前后加上指定的字符
2	–escaped-by	对字段中的双引号加转义符
3	–fields-terminated-by	设定每个字段是以什么符号作为结束，默认为逗号
4	–lines-terminated-by	设定每行记录之间的分隔符，默认是\n
5	–mysql-delimiters	Mysql默认的分隔符设置，字段之间以逗号分隔，行之间以\n分隔，默认转义符是\，字段值以单引号包裹。
6	–optionally-enclosed-by	给带有双引号或单引号的字段值前后加上指定字符。

公用参数：export

序号	参数	说明
1	–input-enclosed-by	对字段值前后加上指定字符
2	–input-escaped-by	对含有转移符的字段做转义处理
3	–input-fields-terminated-by	字段之间的分隔符
4	–input-lines-terminated-by	行之间的分隔符
5	–input-optionally-enclosed-by	给带有双引号或单引号的字段前后加上指定字符

公用参数：hive

序号	参数	说明
1	–hive-delims-replacement	用自定义的字符串替换掉数据中的\r\n和\013 \010等字符
2	–hive-drop-import-delims	在导入数据到hive时，去掉数据中的\r\n\013\010这样的字符
3	–map-column-hive	生成hive表时，可以更改生成字段的数据类型
4	–hive-partition-key	创建分区，后面直接跟分区名，分区字段的默认类型为string
5	–hive-partition-value	导入数据时，指定某个分区的值
6	–hive-home	hive的安装目录，可以通过该参数覆盖之前默认配置的目录
7	–hive-import	将数据从关系数据库中导入到hive表中
8	–hive-overwrite	覆盖掉在hive表中已经存在的数据
9	–create-hive-table	默认是false，即，如果目标表已经存在了，那么创建任务失败。
10	–hive-table	后面接要创建的hive表,默认使用MySQL的表名
11	–table	指定关系数据库的表名

import命令及参数

序号	参数	说明
1	–append	将数据追加到HDFS中已经存在的DataSet中，如果使用该参数，sqoop会把数据先导入到临时文件目录，再合并。
2	–as-avrodatafile	将数据导入到一个Avro数据文件中
3	–as-sequencefile	将数据导入到一个sequence文件中
4	–as-textfile	将数据导入到一个普通文本文件中（默认是as-textfile）
5	–boundary-query 边界查询，导入的数据为该参数的值（一条sql语句）所执行的结果区间内的数据。
6	–columns <col1, col2, col3>	指定要导入的字段
7	–direct	直接导入模式，使用的是关系数据库自带的导入导出工具，以便加快导入导出过程。
8	–direct-split-size	在使用上面direct直接导入的基础上，对导入的流按字节分块，即达到该阈值就产生一个新的文件
9	–inline-lob-limit	设定大对象数据类型的最大值
10	–m或–num-mappers	启动N个map来并行导入数据，默认4个。
11	–query或–e	将查询结果的数据导入，使用时必须伴随参–target-dir，–hive-table，如果查询中有where条件，则条件后必须加上$CONDITIONS关键字
12	–split-by	按照某一列来切分表的工作单元，不能与–autoreset-to-one-mapper连用（请参考官方文档）
13	–table	关系数据库的表名
14	–target-dir	指定HDFS路径
15	–warehouse-dir	与14参数不能同时使用，导入数据到HDFS时指定的目录
16	–where	从关系数据库导入数据时的查询条件
17	–z或–compress	允许压缩
18	–compression-codec	指定hadoop压缩编码类，默认为gzip(Use Hadoop codec default gzip)
19	–null-string	string类型的列如果null，替换为指定字符串
20	–null-non-string	非string类型的列如果null，替换为指定字符串
21	–incremental	mode：append或lastmodified
22	–last-value	指定某一个值，用于标记增量导入的位置
23	–check-column	指定作为增量导入判断的列名

export命令及参数

序号	参数	说明
1	–direct	利用数据库自带的导入导出工具，以便于提高效率
2	–export-dir	存放数据的HDFS的源目录
3	-m或–num-mappers	启动N个map来并行导入数据，默认4个
4	–table	指定导出到哪个RDBMS中的表
5	–update-key	对某一列的字段进行更新操作
6	–update-mode	updateonly或allowinsert(默认)
7	–input-null-string	请参考import该类似参数说明
8	–input-null-non-string	请参考import该类似参数说明
9	–staging-table	创建一张临时表，用于存放所有事务的结果，然后将所有事务结果一次性导入到目标表中，防止错误。
10	–clear-staging-table	如果第9个参数非空，则可以在导出操作执行前，清空临时事务结果表

java代码操作Linux命令

需要下载jar包，sshxcute.jar，将jar包导入java项目
示例代码：
完成通过java代码在Linux中向文件并插入内容

import net.neoremind.sshxcute.core.ConnBean;
import net.neoremind.sshxcute.core.SSHExec;
import net.neoremind.sshxcute.exception.TaskExecFailException;
import net.neoremind.sshxcute.task.impl.ExecCommand;

public class MyXcute
{
    public static void main(String[] args) throws TaskExecFailException
    {
        SSHExec exec=SSHExec.getInstance(new ConnBean("192.168.137.120","root","123456"));

        exec.connect();
        exec.exec(new ExecCommand("echo 'hello world' >> /export/servers/helloworld.txt"));
        exec.disconnect();
    }
}