钉钉支持消息推送和钉钉上查看ABI报表资源。支持任务流调度管理、实时调度管理、变更检测管理、流程管理、计算报表计划任务、用户注册等消息通知推送到钉钉。同时支持添加ABI报表资源到钉钉，在钉钉上查看ABI报表资源。

2.创建企业应用

创建企业应用，首先需要登录钉钉开放平台。

钉钉开放平台：https://open.dingtalk.com/document （创建钉钉应用）

钉钉管理后台：https://oa.dingtalk.com/#/welcome （在【工作台】-【应用管理】-【应用设置】中可设置员工应用可见范围）

用管理员的账号扫一扫登录钉钉开放平台。已创建成功的钉钉企业应用的首页界面（如下图），记录好此处的CorpId（企业ID）。后续钉钉配置时需要。

在企业号中创建一个应用，点击【应用开发】-【钉钉应用】-【创建应用】-【委托服务商开发】即可。此次我们以“H5微应用”举例。

创建好应用后，点击查看该应用的应用凭证与基础信息，查看并记录 “钉钉应用ID(AgentId)、AppKey和AppSecret”，后续ABI中添加钉钉应用时需要。

创建好应用后，如果需要根据钉钉的手机号在系统中匹配用户，需要在开发配置-权限管理下面开通以下3个权限。

以上钉钉企业应用便创建成功，接下来在ABI中添加钉钉应用。

3.添加钉钉应用

在【系统管理】-【系统设置】-【钉钉配置】中点击【添加钉钉应用】按钮，输入相应的信息项后点击确定，钉钉应用即可添加成功。

a. 钉钉应用名称：用来描述应用名称；

b. 钉钉应用ID(AgentId)：在钉钉【应用开发】下所对应的应用中的【凭证与基础信息】中找到；

c. AppKey：在钉钉【应用开发】下所对应的应用中的【凭证与基础信息】中找到；

d. AppSecret：在钉钉【应用开发】下所对应的应用中的【凭证与基础信息】中找到；

e. 企业ID(CorpId)：在钉钉开放平台的【首页】中找到CorpId；

f. 用户匹配规则：有“按钉钉成员ID匹配“和”按手机号匹配”两种匹配方式。默认是手机号匹配；

g. 匹配字段：下拉的字段为系统用户字段，匹配字段默认显示为TEL_（电话），用户也可以选择其他系统用户字段进行匹配。

如若匹配字段选择“用户代码”的匹配方式，则需要在【机构用户】中创建一个与钉钉绑定电话相同的系统用户ID。匹配完成后，去【钉钉配置】-【应用成员】中查看钉钉成员ID与系统用户是否匹配上。

4.钉钉消息推送

钉钉支持任务流调度管理、实时调度管理、变更检测管理、流程管理、计算报表计划任务、用户注册等消息通知推送到钉钉。具体功能操作，可以参照各功能对应的说明。

推送完成后在【系统管理】-【日志】-【钉钉应用推送日志】中可以查看相关日志。

5.钉钉配置ABI菜单链接

添加ABI报表资源到钉钉，在钉钉上查看ABI报表资源。

5.1 菜单链接添加ABI资源

点击钉钉应用【菜单链接】按钮进入到菜单链接页面，有移动端和PC端两种资源链接。

添加移动端的资源链接，点击【添加】按钮，添加移动应用中的资源（如下图）

此需要强调一下，登录亿信ABI时，一定要用外网域名地址访问，并且需要在【系统管理】-【参数配置】-【服务器地址配置】中配置外网地址。不然会提示您：

添加完成后如下， 可以看到资源类型总共分为5种类型：

· 移动应用首页，显示ABI中所有的移动应用资源；

· 资源列表，显示所选择的应用下的所有资源包括：移动应用、公告、待办事项、发起流程；

· 分组资源，显示所选择的分组下的所有资源；

· 表单资源，直接展示该表单；

· 分析表资源，直接展示该分析表。

添加PC端的资源链接，点击【添加】按钮，添加PC端应用中的资源（如下图）

添加完成后如下， 可以看到资源类型总共分为4种类型：

· PC首页，显示PC登录页；

· 表单资源，显示所选择的应用下的表单资源；

· 分析资源，显示所选择的应用下的报表资源；

· 门户资源，直接展示门户资源。

5.2 配置ABI访问菜单

资源的链接地址获取后，在钉钉开放平台【应用开发】-【钉钉应用】中，点击已创建好的应用，在【网页应用】中，添加我们要访问的移动端和PC端的ABI链接，保存后发布成功即可。

5.3 钉钉访问ABI菜单

登录钉钉账号，点击【工作台】，找到对应的应用就可以查看表单或ABI分析表。如下图效果展示。

PC端：

移动端：

备注：
当前登录的钉钉用户需要有能匹配上的系统用户账号，点击挂载了ABI链接的应用后，系统自动用匹配上的系统用户账号免登录ABI来查看报表资源，并且该系统用户账号需要有对应的移动应用或者分析表、表单的查看等权限。

实时JSON解析组件

1.应用背景

实时json解析组件可以从前置实时组件中读取到json格式的数据，然后将其解析成结构化数据，并向后置组件输出。

客户的一个接口返回的用户信息，姓名性别身份证家庭住址等等，这些信息是以JSON的形式返回的，可以通过时JSON解析组件将用户的信息落地到数据库表。 

边缘采集组件

1.组件介绍

本章节主要介绍了如何使用边缘采集输入输出组件实时采集服务器上的日志数据。

前提条件

1、需要提前在采集服务器上安装边缘节点程序。

2、在使用边缘采集输入输出组件前，需要配置边缘节点的IP地址和端口号。具体配置方法可参照边缘节点管理。

目前边缘采集输入组件支持以下五种

（1）边缘采集Avro输入组件：可通过RPC接收Avro数据，将数据输入到内存中。

（2）边缘采集文件目录输入组件：可监听目录下的新文件，将数据输入到内存中，不支持断点续传。

（3）边缘采集Kafka输入组件：可从Kafka中读取数据，将数据输入到内存中。

（4）边缘采集文本输入组件：可监听目录或文件，将数据输入到内存中，支持断点续传。

（5）边缘采集Http输入组件：可接收外部HTTP客户端发送过来的数据，将数据输入到内存中。

边缘采集输出组件支持以下四种

（1）边缘采集HDFS输出组件：可从内存中接收FLUME编译器输入组件的数据，将数据通过RPC实现端到端的批量压缩数据传输。

（2）边缘采集Avro输出组件：可从内存中接收FLUME编译器输入组件的数据，将数据写入HDFS文件系统。

（3）边缘采集HBase输出组件：可从内存中接收FLUME编译器输入组件的数据，将数据写入Hbase时序数据库中。

（4）边缘采集Kafka输出组件：可从内存中接收FLUME编译器输入组件的数据，将数据写入Kafka指定主题中。

2.使用场景

Story1：客户需要采集tomcat服务器实时产生的日志数据，并实时解析入库；为实时平台操作分析提供数据支撑。

操作步骤

1、边缘采集输入输出组件主要是实现了基于服务器应用日志进行实时更新，实时监听服务器的日志数据，当日志数据中有新增后，将新增的数据输出至内存区进行缓存，然后输出到目标中。系统操作流程如下：

操作入口：任务管理-新建实时任务-实时任务设计器 

① 在ETL任务设计器分组“边缘采集输入组件”下拖出“边缘采集Avro输入组件”。双击该组件，配置采集的边缘节点、输入源的具体信息。

a.边缘节点（默认使用内存方式缓存数据）：选择系统设置-数据整合配置-边缘节点，系统根据配置的边缘节点程序，采集服务器上实时产生的日志数据。

b.b.主机名：监听主机名/IP

c.端口：绑定监听端口，该端口需未被占用

d:高级属性：扩展属性用来配置非必填项的其他属性，格式为：key=value。

②在ETL任务设计器分组“边缘采集输出组件”下拖出“边缘采集Avro输出组件”。，将之前配置好的“边缘采集Avro输入组件”连接此组件，然后双击“边缘采集Avro输出组件”，配置数据的输出目标信息。

a.主机名：绑定的主机名/IP

b.端口：监听端口

c:高级属性：扩展属性用来配置非必填项的其他属性，格式为：key=value。

③点击【运行】，系统跟据配置的边缘节点程序，即可实现实时采集服务器上产生的日志数据。

3.注意事项

1、实时任务点击【运行】后，默认在后台一直运行，直到用户点击【取消】后才会终止。

2、采集组件和推送组件支持多对多的连线。

3、边缘采集组件仅支持相互之间进行使用。 

实时CDC日志单表同步组件

1.前置步骤

1.需要提前对原数据库的环境进行操作配置。目前支持的五种数据库。分别为ORACLR,MYSQL,SQLSERVER,POSTGRESQL以及达梦数据库。详情参考数据库表配置。

2.在进行CDC同步的过程中原表必须拥有主键，否则无法达到数据修改和删除同步的效果。

3.服务器中需部署大数据实时处理服务。

2.使用步骤

1.实时CDC日志增量组件，使用主要是通过对数据库日志的监控来进行数据的实时传输。

2.用户通过界面填写必要的配置信息，选择对应的操作便可达到实时数据同步处理的效果。

3.操作生成的事件。

快照：当第一次监控这个数据库表的时候，会对这个数据库表进行一次快照处理。事件都是一条一条数据处理的。

INSERT(插入)：插入分为批量插入和单条插入。无论是批量插入还是单条插入，每个事件也都是按单条数据处理  。

UPDATE(更新)：更新分为批量更新和单条更新。无论是批量更新还是单条更新，每个事件也都是按单条数据处理。

DELETE(删除)：删除也是如此。

1）选择对应的cdc 组件

2）选择需要进行CDC的库表

双击实时CDC日志增量组件，进行属性编辑，选择支持的数据库和表。（选择的数据库和表都应该是经过前提配置开启CDC日志监控的。）

业务操作：即设置同步的操作。默认勾选的三个INSERT\UPDATE\DELETE。

抽取存量数据：1、如果勾选在第一次执行ETL过程时，会同步源表数据更新目标表,之后实时的同步日志增删改数据。先全量，再实时增量。2、不勾选,在第一次执行时开始同步日志得增删改数据.实时增量。

扩展属性设置：提供一个扩展属性设置供用户添加扩展功能。

配置完成后即可运行该任务。

3.运行场景

实时CDC 日志增量组件作为输入组件可后接其他数据处理组件和输出组件。通常我们以CDC作为输入端，后接数据处理组件，然后将处理后的数据推送到kafka中。

此处我们以学生表为例。

表中有ID、NAME、SEX三个字段，其中ID字段设置为主键。

选择完成后，我们在该组件后接过滤组件，将性别（SEX）为空的数据过滤掉，然后将过滤后的数据推送到kafka服务器中。

配置完成后，点击运行任务，等待该任务成功启动运行后，我们可以通过SQL对xsb 进行数据的增删改的操作。被编辑后的数据将会被监控到且推送到kafka服务器中。

我们可通过检查运行过程中的生成的临时表（EDI_CDC_表名），通过查看临时表的数据来检查我们的操作是否被监控到。

其中EDI_CDC_OP代表数据的操作 ，C代表添加数据，D代表删除数据，U代表修改的数据。EDI_CDC_INDEX代表监控操作次数。

4.开启CDC日志方法

4.1oracle

官方文档说明只验证过oracle 12c以及19c。本文档中用oracle 11g验证通过

1 oracle配置

本质是通过oracle归档日志来实现同步的。

1.1 开启归档模式

用sqlplus执行如下命令

sqlplus / as sysdba

connect sys/admin AS SYSDBA

alter system set db_recovery_file_dest_size = 10G;

alter system set db_recovery_file_dest = 'D:\oracle11gr2\oradata\recovery_area'

scope=spfile;

shutdown immediate

startup mount

alter database archivelog;

alter database open;

-- Should now "Database log mode: Archive Mode"

archive log list

exit;

在数据库级别启用最小补充日志记录，并且可以按如下方式配置。

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

★创建表空间，监控已经存在的表空间则不需要创建

CREATE TABLESPACE logminer_tbs DATAFILE

'D:\oracle11gr2\oradata\orcl\logminer_tbs.dbf'

SIZE 25M REUSE AUTOEXTEND ON MAXSIZE UNLIMITED;

★创建user

CREATE USER dbzuser IDENTIFIED BY dbz

DEFAULT TABLESPACE logminer_tbs

QUOTA UNLIMITED ON logminer_tbs

★给用户授权

GRANT CREATE SESSION TO dbzuser;

GRANT SET CONTAINER TO dbzuser;

GRANT SELECT ON V_$DATABASE to dbzuser;

GRANT FLASHBACK ANY TABLE TO dbzuser;

GRANT SELECT ANY TABLE TO dbzuser;

GRANT SELECT_CATALOG_ROLE TO dbzuser;

GRANT EXECUTE_CATALOG_ROLE TO dbzuser;

GRANT SELECT ANY TRANSACTION TO dbzuser;

GRANT LOGMINING TO dbzuser;

GRANT CREATE TABLE TO dbzuser;

GRANT LOCK ANY TABLE TO dbzuser;

GRANT ALTER ANY TABLE TO dbzuser;

GRANT CREATE SEQUENCE TO dbzuser;

GRANT EXECUTE ON DBMS_LOGMNR TO dbzuser;

GRANT EXECUTE ON DBMS_LOGMNR_D TO dbzuser;

GRANT SELECT ON V_$LOG TO dbzuser;

GRANT SELECT ON V_$LOG_HISTORY TO dbzuser;

GRANT SELECT ON V_$LOGMNR_LOGS TO dbzuser;

GRANT SELECT ON V_$LOGMNR_CONTENTS TO dbzuser;

GRANT SELECT ON V_$LOGMNR_PARAMETERS TO dbzuser;

GRANT SELECT ON V_$LOGFILE TO dbzuser;

GRANT SELECT ON V_$ARCHIVED_LOG TO dbzuser;

GRANT SELECT ON V_$ARCHIVE_DEST_STATUS TO dbzuser;

exit

★创建表并授权

sqlplus / as sysdba

conn dbzuser/dbz;

CREATE TABLE STU ( "s_id" INT PRIMARY KEY, "s_name" VARCHAR ( 255 ) ); //

ALTER TABLE DBZUSER.STU ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

4.2Pg

★支持版本

postgresql支持9.4以上版本

★ 准备步骤

默认已经安装了pg数据库

源码目录：/opt/postgresql-12.8

★安装解码插件

wal2json ：wal2json组织维护的插件，用于将日志转换成json。

★安装wal2json

1.从githhub：  下载最新的稳定版的源代码包 https://github.com/eulerto/wal2json/tree/wal2json_2_4

测试中使用的是2.4版本。

2 解压缩wal2json插件到指定目录

执行命令：unzip wal2json-wal2json_2_4.zip

进入wal2json的目录： cd wal2json-wal2json_2_4

3 编译打包，并安装，执行如下命令

make

make install

★ 配置postgresql.conf

打开postgresql安装目录data文件夹下的postgresql.conf文件，增加一行

shared_preload_libraries = 'wal2json'。

★配置replication slot

Replication slot和connector是一对一的关系，每启动一个CDC同步任务就需要一个复制槽。且复制槽还和快照有关，删除了复制槽则需要重新快照,如果复制槽不够，或者复制槽名称相同都会报错，通过postgresql.conf配置复制槽的数量,设置如下配置：

max_wal_senders = 10   #wal日志的sender

max_replication_slots = 10  #总的复制槽数量

保存后重启postgresql

★停止和启动postgresql

1 切换到pg的超级用户下： su - postgres

2停止：

/usr/local/postgresql/bin/pg_ctl -D /usr/local/postgresql/data -l

/usr/local/postgresql/data/logfile stop

3 启动：

/usr/local/postgresql/bin/pg_ctl -D /usr/local/postgresql/data -l

/usr/local/postgresql/data/logfile start

进入pg sql

su - postgres

psql -U postgres

★ 配置权限

1 创建一个拥有复制和登录权限的角色，角色名自行命名，本例中是dbz_role

tips:使用管理员账号也是可以的，管理员(postgres)默认拥有这些权限 dbz_role   LOGIN; CREATE ROLE REPLICATION

2 将dbz_role这个角色赋予给需要监听的数据库用户,超级管理员账号则忽略

grant dbz_role dbz_user

★ 配置pg_hba.conf

打开pg_hba.conf这个文件，在文件的最后增加一行，dbz_user是数据源里设置的用户名。也可以使用postgres超级账号。ip是数据工厂所在的服务器地址如果数据工厂是集群环境，则配置多条，每条对应不同的ip。配置好后重启postgresql.

host  replication   dbz_user    172.21.1.7/32      trust

4.3Sqlserver

SQL server数据库配置

基于sqlserver的cdc特性只支持SQL Server 2016 Service Pack 1 (SP1) 及其以后的标准版或者企业版

★sqlserver环境准备

详情请参考以下链接（linux环境下配置，window环境下配置过于繁琐且卸载不易）

https://blog.csdn.net/yutenys/article/details/122154593?ops_request_misc=%257B%2522request%255Fid%2

522%253A%2522164620290116780255244215%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fall

.%2522%257D&request_id=164620290116780255244215&biz_id=0&utm_medium=distribute.pc_search_result.non

e-task-blog-2~all~first_rank_ecpm_v1~rank_v31_ecpm-1-122154593.pc_search_result_control_group&utm_t

erm=centos7%E5%AE%89%E8%A3%85sqlserver2019&spm=1018.2226.3001.4187

步骤2——步骤4

★开启sqlserver的CDC模式

# sqlcmd -S localhost -U sa -PEsen1234             

（sa是默认账号，Esen1234是我设置的密码）

1>create database test

2>go

1>use test

2>go

★开启test库的CDC

1>EXEC sys.sp_cdc_enable_db

2>go

2.2创建文件组

1>use test

2>go

1>alter database test  add filegroup fg1

2>go

★创建文件组下的文件

1> alter database test

2> add file

3> (name='ffggg1',filename='/var/opt/mssql/data/ffggg1.mdf')

4> to filegroup fg1

5> go

★开启sqlserver代理

# /opt/mssql/bin/mssql-conf set sqlagent.enabled true

# systemctl restart mssql-server

# sqlcmd -S localhost -U sa -PEsen1234

1>EXEC sp_configure 'show advanced', 1;

2>RECONFIGURE;

3>EXEC sp_configure 'allow updates', 0;

4>RECONFIGURE;

5>EXEC sp_configure 'Agent XPs', 1;

6>RECONFIGURE;

7>go

★开启test库的t1表的CDC

1>use test

2>go

1>EXEC sys.sp_cdc_enable_table

2>@source_schema = N'dbo',

3>@source_name   = N't1',

4>@role_name     = N'NULL',

5>@filegroup_name = N'fg1',

6>@supports_net_changes = 0

7>go

★ 修改表结构后重新cdc（重点关注）

修改源表表结构，需要重新进行cdc配置才可以，分两步

第一步，先禁用源表之前的cdc配置，这样会删除源表对应的cdc数据库表以及一些配置信息

EXEC sys.sp_cdc_disable_table @source_schema = 'dbo', @source_name = 't1',

@capture_instance = 'dbo_t1';

go

其中捕获实例capture_instance 这个参数的值默认是“模式 + 下划线 + 表名“

第二步 重新执行启用表cdc的存储过程

EXEC sys.sp_cdc_enable_table

@source_schema = N'dbo',

@source_name = N't1',

@role_name = N'NULL',

@filegroup_name = N'fg1',

@supports_net_changes = 0

go

★ 注意事项

1 创建文件组后一定要创建对应的文件

2 不支持sql server中的timestamp类型，因为这个类型在sql server中和时间毫无关系

4.4Mysql

本文档中用mysql5.7和mysql8.0验证通过

★ mysql配置

本质是通过mysql binlog日志来实现同步的。

5.3.1 开启binlog日志

（1）检查是否开启binlog日志

mysql> show global variables like 'log_bin';

返回结果

ON表示已开启binlog日志，如果是OFF则没有开启。

（2）如果没开启，需在/etc/my.cnf中加入以下内容

server_id=2

log_bin = mysql-bin

binlog_format = ROW

expire_logs_days = 30

（3）重启mysql

systemctl restart mysqld

5.3.2

//创建数据库

create database test_db default character set utf8 default collate

utf8_general_ci;

//创建user

create user 'test_user'@'%' identified by '123456';

//授权

grant all privileges on test_db.* to 'test_user'@'%';

//此类权限不是针对某个数据库的，因此需使用on *.*来进行授权

grant SELECT,RELOAD,SHOW DATABASES,REPLICATION SLAVE,REPLICATION CLIENT on *.* to

'test_user'@'%';

//刷新权限

flush privileges;

//退出

exit;  

目前加密级组件支持AES/ECB、AES/CBC、SM2、SEM4等多种算法。下图所示：

【操作步骤】

操作入口：【任务管理>任务定义>新建-实时任务>输出-表流式输出组件】

（1）新建表流式输出组件

打开任务编辑器，左侧组件面板中找到转换分组栏，选择表流式输出组件拖拽到右边编辑区域。

（2）界面设置

双击表流式输出组件，打开目标设置界面，设置连接池、库表等，对字段进行映射、拾取、删除操作。同时，勾选“关键字”后，支持更新插入

切换到附加设置界面，选择是否忽略异常、参考示例填写扩展属性，点击确定。注：表流式输出组件支持关系型数据库、Hive等，可以选择对应的连接池和表，选择表后自动显示表字段，并支持表不存在则创建，支持表更新方式和批量大小设置，支持扩展属性。 

注意：表流式输出组件支持将字段注释一起输出到目标表中。具体方式可参考表输出。

【操作步骤】

1. 前置步骤：

1.1需部署大数据集群环境，详情见部署手册。此处以xxx.xxx.xxx.3这个大数据环境为例。

1.2.将开源的spark.zip和hadoop.tar.gz（联系项目人员获取）解压到睿治服务器的工作目录下，解压后先删除hadoop/etc/hadoop目录下的所有文件。

1.3.解压后将需连接的大数据集群服务器中edi-RTPS-el7/installdir/hadoop-2.7.4/etc/hadoop目录中的hdfs-site.xml、core-site.xml、yarn-site.xml文件放置到睿治工作目录的 hadoop/etc/hadoop 目录下。

1.环境配置

1.1 local模式参数配置

（1）添加配置文件

HADOOP_HOME/hadoop/etc/hadoop_local目录 或 睿治工作目录/hadoop/etc/hadoop_local目录 放入以下文件（如果配置了HADOOP_HOME，配置文件需优先放入该目录下，后续不再说明）

hbase-site.xml(从需要使用的hbase库的数据库驱动中获取)

（2）放依赖包

睿治工作目录/spark/jars目录下放入数据质量所有依赖包

界面配置

1.2 standalone模式参数配置

（1）添加配置文件

睿治工作目录/hadoop/etc/hadoop目录 放入以下文件hbase-site.xml(从需要使用的hbase库的数据库驱动中获取)

（2）上传依赖包

睿治服务器spark所在路径下的jars目录中需要添加所有依赖包

hdfs://xxx.xxx.xxx.3:8020/edi/spark放入edi-job-spark.jar包

hdfs://xxx.xxx.xxx.3:8020/edq/spark放入edq-spark.jar包

spark集群所有服务器执行以下操作：

/opt/edi/spark/depend_jars目录下放入数据工厂所有依赖包

/opt/edq/spark/depend_jars目录下放入数据质量所有依赖包

（3）界面配置

参数配置：

1.3 Yam+sample模式参数配置

（1）添加配置文件

睿治工作目录/hadoop/etc/hadoop目录 放入以下文件hbase-site.xml(从需要使用的hbase库的数据库驱动中获取)

（2）上传数据质量、数据工厂的依赖包到hdfs上

hdfs://xxx.xxx.xxx.182:8020/edi/spark放入edi-job-spark.jar包

hdfs://xxx.xxx.xxx.182:8020/edi/spark/depend_jars放入数据工厂依赖包

hdfs://xxx.xxx.xxx.182:8020/edq/spark放入edq-spark.jar包

hdfs://xxx.xxx.xxx.182:8020/edq/spark/depend_jars放入数据质量依赖包

（3）界面配置

参数配置：

数据质量模块：

edq.spark.executor.instances=5

edq.spark.executor.memory=4g

edq.spark.executor.cores=2

edq.spark.driver.memory=2g

edq.spark.driver.maxResultSize=4g   这些为大数据质检的资源配置

数据集成模块:

单体：

spark.driver.host=xxx.xxx.xxx.27  数据集成实时任务运行的配置其中xxx.xxx.xxx.27为睿治服务器的ip地址

微服务集群：

edi.stream.spark.driver.host=$MY_HOST_IP

edi.stream.spark.driver.port=19001

edi.stream.spark.driver.blockManager.port=19002

edi.stream.spark.driver.bindAddress=0.0.0.0

其中$MY_HOST_IP为动态ip地址,需要在容器云平台中的数据工厂的服务中配置

edi.stream.spark.driver.port=19001

edi.stream.spark.driver.blockManager.port=19002

这两个端口需要在数据工厂服务中进行添加

1.4 Yam+kerbors模式参数配置

前置步骤：在带kerbors认证的服务器上部署大数据环境，这里以xxx.xxx.xxx.216为例

（1）添加配置文件

睿治工作目录/hadoop/etc/hadoop目录 放入以下文件hbase-site.xml(从需要使用的hbase库的数据库驱动中获取)

（2）spark集群所有服务器上传kerberos文件

/opt/bigdataconf/spark/kerberos目录下放以下文件

krb5.conf、hive.service.keytab

krb5.conf从带kerbors认证的大数据服务器中获取，获取路径/etc/krb5.conf

service.keytab从带kerbors认证的大数据集群服务器中获取，获取路径/etc/security/keytabs/hive.service.keytab

（3）睿治任意服务器放入以下文件（如/opt）

krb5.conf、hive.service.keytab

（4）上传数据质量、数据工厂的依赖包到hdfs上

hdfs://xxx.xxx.xxx.216:8020/edi/spark放入edi-job-spark.jar包

hdfs://xxx.xxx.xxx.216:8020/edi/spark/depend_jars放入数据工厂依赖包

hdfs://xxx.xxx.xxx.216:8020/edq/spark放入edq-spark.jar包

hdfs://xxx.xxx.xxx.216:8020/edq/spark/depend_jars放入数据质量依赖包

（5）

其中hive/petabase216.esen.com@EXAMPLE.COM中 hive可根据我们选择的 hive.service.keytab文件进行变换，@EXAMPLE.COM指定后缀的编写方式可从hive.service.keytab文件中获取。

参数配置：

1.5 实时处理服务支持本地化配置

2.注意事项

1.数据质量spark执行参数详解（仅作用于大数据质检）：

edq.spark.executor.instances:Spark 应用程序中执行器的数量。

edq.spark.driver.memory:spark驱动程序（driver）的内存分配的大小。

edq.spark.executor.memory:Spark 应用程序中每个执行器的的内存。

edq.spark.driver.maxResultSize:限制驱动程序（driver）接收和返回的结果集大小。

edq.spark.executor.cores:每个执行器（executor）使用的 CPU 核心数。

2.数据工厂sprak执行参数详解

实时任务资源配置参数（仅作用实时任务）：

edi.stream.spark.num.executors:控制Spark 应用程序中执行器的数量。

edi.stream.spark.executor.memory:指定每个执行器（executor）所分配的内存量。

edi.stream.spark.executor.cores:指定每个执行器使用的 CPU 核心数。

edi.stream.spark.cores.max：指定应用程序可以使用的最大 CPU 核心数量。

在资源充足的情况下，我们可通过增大这些参数的配置来提高大数据质检和ETL运行的速率。

3. 依赖jar包获取方式

在使用一键部署大数据环境安装包后，我们可以从部署安装的HDFS服务器中获取，具体路径为:/bigdataconf/spark/depend_jars。具体依赖jar包名（联系项目人员获取）。部分数据库驱动依赖jar包需手动添加。

Mongodb输入组件可直接从组件中配置mongodb数据源，选择集合进行数据的抽取，作为后续组件的数据来源。

该组件输入设置tab页主要配置分别是mongodb数据源链接地址、用户名、密码。配置好对应的数据库链接信息后点击联通检测，即可检测数据库联通情况。（睿治273 版本新增mongodb数据源，可在数据源处直接建立mongodb数据源连接信息，组件中选择即可。）

注：由于mongodb组件是在组件中直接配置链接信息，所以mongodb的数据库驱动需要手动得根据客户使用实际版本放在工作目录下指定路径下，路径为/workdir/lib/edi/mongodb.

配置完数据源连接配置联通后，可在字段列表页面选择指定集合，读取其中字段属性。 此外，该页面支持配置数据过滤条件，用于在抽取时筛选符合要求的记录。此处配置信息与表输入一致。

Mongodb组件支持最大值增量更新。最大值增量更新方式，与上一次记录的字段最大值比对，大于上次记录的最大值认为是最新数据进行输出。最大值增量字段支持选择多个，当选择多个时是取多个字段的并集进行比较。

2.mongodb输出组件

mongodb输出组件用于将前置组件获取到的结果集，加载到mongodb数据库的指定的集合中。

Mongodb输出组件与输入组件一致，也是在组件内配置mongodb数据源的链接信息。同样由于mongodb组件是在组件内配置链接信息的，所以使用前需要在工作目录中存放实际使用版本的mongodb数据库驱动。具体存放地址见mmongodb输入组件。

字段列表页面可选择集合，集合存在的情况下会自动读取集合字段属性，当集合不存在时会自动读取前置组件的字段属性。勾选不存在则创建，执行任务后回在目标数据源创建对应的集合。

Mongodb输出组件支持数据覆盖、数据附加、更新插入、数据更新等多种更新方式。进行数据更新、更新插入时需要设置关键字，且集合内数据字段结构需保持一致。

本文档是ETL任务的入门示例，希望大家通过这个示例理解ETL任务的制作步骤以及设计思路。下文我们以个人周报数据抽取为例，讲述ETL任务的制作步骤。

具体需求如下：

1)【个人周报表IGRZB_B】中去掉“是否合计行”为1的数据，合计行为1代表是合计数据，我们需要保留明细数据，去掉合计行；去掉“参与任务”为空的垃圾数据，个人周报填写必须包含参与的任务。

2)完善个人周报信息，将【员工信息表FACT_YGXX】中的“所属部门”、“工作地点”等维度字段添加到个人周报信息中。

3)“参与任务”由任务编号+员工id组成，员工id位数固定为7位，例如01002.01060020007P000.1400003，01002.01060020007P000为任务编号，1400003为员工id。需要截掉后面的员工id，保留参与任务编号。

4)为不影响源表数据，需要把处理好的数据抽取到一张新数据表中。

附件:

图1个人周报表表结构IGRZB_B

图2 个人周报表部分数据

2. 制作步骤

2.1新建ETL任务

打开设计区的ETL任务列表，点击【新建】，选择任务类型，进入ETL任务设计器。

2.2添加表输入组件

根据需求，我们要对数据库中的表进行ETL，将IGRZB_B作为输入表。

【表输入】组件支持选择任意接入的连接池中的数据库表作为输入表。

将表输入组件拖入到右边编辑区，双击组件进行配置，选择连接池和表。我们选择个人周报表IGRZB_B。

2.3添加过滤组件，实现数据过滤

根据需求，我们需要去掉“是否合计行”为1和“参与任务”为空的脏数据。

【过滤组件】主要完成过滤功能，在输入数据中筛选出需要的结果集合，最后装载到目标中去，类似于数据库的where操作。

拖入过滤组件，鼠标悬浮到前置任务组件中间位置，会出现一个黄色箭头的圆点，并且鼠标会由指针变为手状，鼠标拖动这个圆点向右移动，这时黄色箭头也会跟着向右移动，然后将鼠标落在需要连接的组件中央松开鼠标即可完成两个任务组件的连接。

双击过滤组件，在过滤组件的公式编辑区域可以编辑过滤表达式，完成对输入结果的过滤。可以选择简单过滤或自定义过滤，这两种过滤方式产生的效果是一样的，这里我们选择简单过滤。点击【选择】按钮选择需要过滤的字段“ISHJ_(是否合计)”和“CYRW(参与任务)”，选择操作符、取值，以及条件之间的连接符，配置如下图所示：

如果使用自定义过滤，则如下图所示

2.4添加连接组件，完成两表连接

按照需求，我们需要将【员工信息表FACT_YGXX】中的“所属部门”、“工作地点”等维度字段添加到个人周报，此时需要将【员工信息表FACT_YGXX】也作为输入表，并与【个人周报表IGRZB_B】完成表连接。

【连接组件】用于完成将多个表进行连接，将连接后的结果输出。连接组件的连接方式主要有以下几种：内连接、左连接、右连接、全连接，与数据库里表连接的方式类似，通过确定关联字段和关联条件，最后得到匹配连接后的结果集。连接的输入是两个以上，输出只有一个。

拖入连接组件，过滤组件和【员工信息表FACT_YGXX】的表输入组件作为连接组件的输入，双击连接组件进行配置。根据需求员工信息表是作为补充，所以选择连接方式为左连接，并设置两表连接条件。

注意：如果连接组件的输入表中有相同字段，系统会自动创建连接表达式，如不需要可以自行删减连接条件。

2.5添加表达式组件，对字段进行处理

【表达式组件】可以通过表达式对前置组件中各字段的数据进行转换处理，得到新的结果集数据。

拖入表达式组件，作为连接组件的输出。双击表达式进行配置，在弹出框配置需要处理的字段，点击表达式框的三个点（见下图1），在弹出框中编辑处理逻辑。可以使用产品内置函数，也可以使用数据库内置的函数。

按照需求，我们需要截取掉“参与任务”字段中的员工id，保留参与任务编号，表达式应写为：

LEFT(Join1.CYRW,LENGTH(Join1.CYRW)-8)

其中：

LENGTH(Join1.CYRW)-8，是获取参与任务的数据长度，减去人员编码固定7位数和“.”的1位。

至此，数据处理部分已经全部完成，最后需要将处理好的结果集输出到一张新的数据库表中。

2.6添加表输出组件

以上我们数据就处理好了，现在需要将结果集输出到一张新的数据库表中。拖入【表输出】组件，作为表达式组件的输出，双击表输出组件，选择要输出的数据源和表，并配置“字段映射”。如果目标表在连接池中不存在，可以勾选“不存在则创建”，此选项也可以避免将来输出表被删除后，ETL无法正常执行。然后在目标设置中选择更新方式和批量大小。

至此，ETL任务就设计完成了，得到如下ETL任务。

2.7运行调试

ETL任务设计完成后，点击运行，可以执行ETL任务。这里我们需要注意的是：ETL任务最好不要全部设计完成后点击运行，要在设计过程中逐步对各个组件的运行结果进行检查调试。系统提供逐步调试的功能，在想要预览结果的组件上，单击鼠标右键，选择“运行到”并在弹出窗口点确认，ETL任务就会只执行到选择的组件。例如，选择执行到“过滤组件1”，如下图所示。

表示已经执行过的组件，表示当前运行到的组件，表示还未执行的组件。继续执行剩余全部组件，停止当前执行。

选择已经执行或当前运行到的组件，单击右键，点击“预览数据”可预览当前处理逻辑的执行结果。可查看当前生成的SQL是否符合需求。

2.8保存ETL任务

最后点击菜单栏中【保存】按钮，在弹出的保存对话框中设置ETL任务的名称和标题即可完成保存，如下图：