业务背景

公司上大数据，要把sqlserver里的业务数据实时同步到大数据平台上。几天调研后选择StreamSet作为ETL工具。技术选型的理由主要有几点：

1. sqlserver的坑太深，网上找了很多工具对sqlserver的支持力度都不是很大(微软全家桶的要哭了~)
2. 自己开发ETL程序耗时太长，同时配套的ETL metrics工具也需要配备，劳民伤财。感觉有时间开发不如把精力放在业务数据研究和指标计算上。
3. sqlserver支持两种实时同步机制：CDC和Change Tracking，CDC使用起来比较重 果断放弃。

streamset安装

我用的是CDH6.2版本集成化了StreamSet，方便管理。

1. streamset下载地址：https://archives.streamsets.com/index.html
   
2. 配置本地yum源
   将下载的三剑客文件放在一个新建文件夹中，并移动到/var/www/html 目录中，做离线包的下载地址，用浏览器访问如下，表示成功
   
3. 配置CSD
   将STREAMSETS-3.1.4.0.jar拷贝到/opt/cloudera/csd,并更改权限，然后重启cloudera-scm-server服务，没有csd文件夹就自己新建一个。

4.下载分发激活Parcel包
传送门：https://cloud.tencent.com/developer/article/1078852

sqlserver之Change Tracking

sqlserver上开启Change Tracking，这里要做到表级别Change Tracking

1. 开启库级别Change Tracking

ALTER DATABASE 数据库名 SET CHANGE_TRACKING = ON (CHANGE_RETENTION = 2 DAYS,AUTO_CLEANUP = ON)

2. 开启表级别Change Tracking

ALTER TABLE [dbo].[Department] ENABLE CHANGE_TRACKING WITH (TRACK_COLUMNS_UPDATED = ON)

Streamset之sqlserver配置

1. 右上角点击All Stages找到SQLxxx
   
2. 配置分析
   
   
   
   
   

StreamSet之Hive配置

配置Hive要使用HiveMeData和HiveMeStore两个插件
重要参数详解：

1. Database Expression：Hive数据库选择，不写则是默认库
2. Partition Configuration：设置Hive分区的参数，这里可以设置成多级分区，只要选择“+”就行，其中Partition Value Expression是获取分区号数据的表达式，可以自己生成也可以根据表字段里的数据定义分区。因为我们数据原因，我这里表达式写的很蛋疼~有关表达式的写法建议大家直接去撸下官网(PS:Streamset的官网文档还是很不错的，个人感觉比flink强多了！！)。
   官网functions传送门：https://streamsets.com/documentation/datacollector/latest/help/datacollector/UserGuide/Expression_Language/Functions.html#concept_lhz_pyp_1r
   

StreamSet之HDFS配置

重要参数详解：
File System URI:hdfs的URI连接
Configuration Files Directory：hdfs的config文件位置，需要加载core-site.xml
Files Prefix:生成hdfs的文件前缀，建议大家改一下，不然生成的文件名巨长~
Files Suffix：生成hdfs的文件后缀，为用了txt格式，记住不用加点
Directory in Header：打钩后会直接落盘到hive的/user/hive/warehouse/中，如果不打钩，则可以自定义落盘路径
Compression Codec ：选择压缩算法
Use Roll Attribute ：回滚机制

StreamSet之踩坑实战

处理数据同步过程中避免不了加减字段，清洗数据等问题，业务越复杂的OLTP架构中数据迁移越是麻烦，在学习使用StreamSet中，建议大家先去撸清楚官网。我自己在学习使用中总结两个方面感觉很有必要：

1. 函数表达式，这里的坑很多，要想使用数据，调用数据都要通过表达式来完成
2. Field开头的插件大多是对数据做处理加工的，大家使用前一定要对Field开头的插件挨个看一遍。我自己就使用了Field Type Converter(字段值类型转换)、Expression Evaluator(表达式执行器)

专注大数据生态、Spark/Flink、Java/Scala仔
架构未满、Coding至上、一直在路上

转自：http://www.cnblogs.com/micheng11/archive/2008/07/08/1237905.html

SQL Server 中对已经定义的变量赋值的方式用两种，分别是 SET 和 SELECT。

SQL Server推荐使用 SET 而不是 SELECT 对变量进行赋值。
当表达式返回一个值并对一个变量进行赋值时，推荐使用 SET 方法。

下表列出 SET 与 SELECT 的区别。请特别注意红色部分。

下面以具体示例来说明问题：

create table chinadba1(
userid int ,
addr varchar(128) 
)
go
insert into chinadba1(userid,addr) values(1,'addr1')
insert into chinadba1(userid,addr) values(2,'addr2')
insert into chinadba1(userid,addr) values(3,'addr3')
go

declare @addr varchar(128)
set @addr = (select addr from chinadba1)
/*
--出错信息为
服务器: 消息 512，级别 16，状态 1，行 2
子查询返回的值多于一个。当子查询跟随在 =、!=、<、<=、>、>= 之后，或子查询用作表达式时，这种情况是不允许的。
*/
go

declare @addr varchar(128)
select @addr = addr from chinadba1
print @addr --结果集中最后一个 addr 列的值
--结果: addr3 
go

declare @addr varchar(128)
set @addr = '初始值'
set @addr = (select addr from chinadba1 where userid = 4 )
print @addr --null值 
go

declare @addr varchar(128)
set @addr = '初始值'
select @addr = addr from chinadba1 where userid = 4
print @addr --保持原值
go

需要注意的是，SELECT 也可以将标量子查询的值赋给变量，如果标量子查询不返回值，则变量被置为 null 值。
此时与 使用 SET 赋值是完全相同的
对标量子查询的概念大家应该都觉得陌生，举个例子就能说明

declare @addr varchar(128)
set @addr = '初始值'
--select addr from chinadba1 where userid = 4 为标量子查询语句
select @addr = (select addr from chinadba1 where userid = 4) 
print @addr --null值

go

另外：

用SELECT给变量赋值

使用SELECT的另一种形式也可以获得同样的结果。对变量要在赋值前要先声明。使用SELECT语句来替代SET命令的主要优点是，可以在一个操作内同时给多个变量赋值。执行下面的SELECT语句，通过SELECT语句赋值的变量就可以用于任何操作了。

      在平常的业务逻辑中，当两个对象之间是多对多关系的时候，我们建表一般有两种方法，一种就是建立一个中间表，还有一种就是在一张表中的某一个字段，存另一个对象的id逗号隔开的那种形式。今天我们主要针对第二种方法的一些处理，因为，在mysql中和sqlerver中，针对逗号隔开的形式的处理，两种数据库的处理是不一样的。当然orcal数据库也是不一样的，今天暂时不写orcal数据库的不同。

  准备工作：我们需要先建立两张表。用户表和按钮表（图一cust表 图二按钮表）

下面我们通过做题的形式，来进行理解两种数据库对这两个函数的不同

1.查找出含有审核或者发布按钮的用户

（1）mysql中的解决方法

SELECT *
from  cust
where FIND_IN_SET('4',buttonids) or FIND_IN_SET('5',buttonids)

 （2）sqlserver中的解决方法

SELECT * from cust
where
(buttonids like '%,4' or buttonids like '4,%' or buttonids like '%,4,%' or buttonids = '4') 
  or 
(buttonids like '%,5' or buttonids like '5,%' or buttonids like '%,5,%' or buttonids = '5')

   结果：

 2.查询出用户列表，同时后面跟一列显示该用户所拥有的 按钮名称也是逗号隔开的那种形式           

（1）mysql中的解决方法

SELECT
	ta.custid,
	ta.username,
	GROUP_CONCAT(ta.buttonname) AS buttonnames,
	GROUP_CONCAT(ta.buttonname,'--',ta.buttonid) AS buttonidandnames
FROM
	(
		SELECT
			c.custid,
			c.username,
			c.`password`,
			c.buttonids,
			b.buttonid,
			b.buttonname,
			FIND_IN_SET(b.buttonid, c.buttonids) nums
		FROM
			cust c,
			t_button b
	) ta
WHERE
	ta.nums > 0
GROUP BY
	ta.custid,
	ta.username

（2）sqlserver中的解决方法

SELECT
	c.custid,
	c.username,
	stuff (
		(
			SELECT
				',' + CONVERT (VARCHAR, b.buttonname)
			FROM
				t_button b
			WHERE
				1 = 1
			AND charindex (
				',' + CONVERT (VARCHAR, b.buttonid),
				',' + c.buttonids
			) > 0 FOR XML path ('')
		),
		1,
		1,
		''
	) AS buttonnames,
	stuff (
		(
			SELECT
				',' + CONVERT (VARCHAR, b.buttonname) + '--' + b.buttonid
			FROM
				t_button b
			WHERE
				1 = 1
			AND charindex (
				',' + CONVERT (VARCHAR, b.buttonid),
				',' + c.buttonids
			) > 0 FOR XML path ('')
		),
		1,
		1,
		''
	) AS buttonidandnames
FROM
	cust c

结果：

用到的知识点：

a.find_in_set('a', 列名称)

b.group_contact(列名称)，该函数，应用在group by分组之后的数据。

c.sqlserver中的stuff函数。

补充的知识点：

a.find_in_set(列名称,'a,b,c,d') 类似于 列名称  in （a,b,c,d）;

b.contact(列名称1，列名称2，列名称3)  as  names   将其他列的值连起来

参考链接：

https://blog.csdn.net/qq_35374517/article/details/76687049

https://www.cnblogs.com/xiaoxi/p/5889486.html

https://www.cnblogs.com/blazezzz/p/7359575.html?%20?%20?