客快物流大数据项目(二十五)：初始化业务数据

2022-02-08 10:43:50 浏览数 (1)

目录

初始化业务数据

一、安装OGG源端

1、在Oracle中创建OGG相关的用户和表空间

2、OGG源端初始化

二、配置OGG源端

1、Oracle创建物流相关表

2、配置管理器MGR进程

3、​​​​​​​配置extract进程

4、配置pump进程

5、​​​​​​​配置define文件

三、配置OGG目标端

1、​​​​​​​拷贝源端的define文件到目标端

2、​​​​​​​配置管理器MRG进程

3、配置Replicate进程

4、​​​​​​​配置kafka.props

​​​​​​​5、最后确认所有的进程

四、​​​​​​​OGG测试

1、​​​​​​​启动kafka消费者

2、​​​​​​​archivelog日志路径

初始化业务数据

一、安装OGG源端

1、在Oracle中创建OGG相关的用户和表空间

操作步骤说明

1 创建表空间在磁盘中的物理路径（需要到root用户操作）

mkdir -p /u01/app/oracle/oggdata/orcl/

chown -R oracle:oinstall /u01/app/oracle/oggdata/orcl

2 进入sqlplus

切换到oracle用户：su - oracle

登录sqlplus：sqlplus "/as sysdba"

3 创建oggtbs表空间

CREATE TABLESPACE "TBS_LOGISTICS" DATAFILE '/u01/app/oracle/oradata/orcl/tbs_logistics.dat' SIZE 500M AUTOEXTEND ON NEXT 32M MAXSIZE UNLIMITED EXTENT MANAGEMENT LOCAL;

4 创建ogg用户(用户名和密码都是ogg)

CREATE USER ogg IDENTIFIED BY ogg DEFAULT TABLESPACE TBS_LOGISTICS;

5 赋予ogg用户dba权限

GRANT connect,resource,dba to ogg;

2、OGG源端初始化

操作步骤说明

1 使用oracle用户登录源端OGG的命令行中

su – oracle

cd $OGG_SRC_HOME

./ggsci

2 初始化源端OGG目录

注意：如果不在OGG_SRC_HOME下，初始化OGG目录时会报错

create subdirs

退出OGG命令行客户端：exit

3 检查源端OGG初始化后的目录

初始化完成后，可以查询在$OGG_SRC_HOME下是否存在dirchk、dirdat、dirdef、dirjar、dirout、dirpcs、dirprm、dirrpt、dirsql、dirtmp共11个目录。

二、​​​​​​​配置OGG源端

1、​​​​​​​Oracle创建物流相关表

属性名

属性值

主机地址

192.168.88.10

端口号

1521

数据库实例名称

ORCL

数据库实例类型

Service Name

用户名

ogg

角色

Normal

密码

ogg

JDBC访问URL

jdbc:oracle:thin:@//192.168.88.10:1521:ORCL

JDBC驱动名称

使用OracleDataSource连接池，无需配置Driver

业务SQL

代码语言：javascript复制
-- 创建所有的序列
CREATE SEQUENCE tbl_emp_info_map_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_driver_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_emp_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_tt_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_charge_standard_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_company_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_company_dot_map_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_company_route_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_company_warehouse_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_courier_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_deliver_region_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_delivery_record_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_department_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_fixed_area_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_goods_rack_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_job_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_out_warehouse_dtl_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_pkg_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_postal_standard_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_push_warehouse_dtl_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_service_evaluation_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_store_grid_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_vehicle_monitor_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_rack_map_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_receipt_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_waybill_line_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_waybill_record_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_work_time_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_test_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_areas_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_deliver_package_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_customer_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_codes_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_consumer_address_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_receipt_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_send_vehicle_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_vehicle_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_dot_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_transport_tool_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_dot_transport_tool_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_address_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_route_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_push_warehouse_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_out_warehouse_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_warehouse_emp_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_express_package_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_express_bill_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_consumer_sender_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_collect_package_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;
CREATE SEQUENCE tbl_waybill_id_seq INCREMENT BY 1 START WITH 1 MINVALUE 1 MAXVALUE 999999999999999;

-- 创建所有的业务表
CREATE TABLE "tbl_emp_info_map" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"company_id" NUMBER(19,0), 
"dot_id" NUMBER(19,0), 
"emp_id" NUMBER(19,0), 
"job_id" NUMBER(19,0), 
"dep_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
CONSTRAINT "PK_TBL_EMP_INFO_MAP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS";

CREATE TABLE "tbl_driver" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"job_number" NVARCHAR2(50), 
"name" NVARCHAR2(50), 
"gender" NVARCHAR2(100),
"birathday" DATE, 
"state" NUMBER(19,0), 
"driver_license_number" NVARCHAR2(100), 
"driver_license_type" NUMBER(19,0), 
"get_driver_license_dt" DATE, 
"car_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DRIVER" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_emp" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"emp_number" NVARCHAR2(50), 
"emp_name" NVARCHAR2(50), 
"emp_gender" NUMBER(10,0), 
"emp_birathday" DATE, 
"state" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_EMP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_transport_tool" (  
"id" NUMBER(19,0) NOT NULL ENABLE, 
"warehouse_id" NUMBER(19,0), 
"transport_tool_id" NUMBER(19,0), 
"allocate_dt" DATE, 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_WAREHOUSE_TRANSPORT_TOOL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_charge_standard" (   
"id" NUMBER(19,0) NOT NULL ENABLE, 
"start_area_id" NUMBER(19,0), 
"stop_area_id" NUMBER(19,0), 
"first_weight_charge" NUMBER(19,0), 
"follow_up_weight_charge" NUMBER(19,0), 
"prescription" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_CHARGE_STANDARD" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS";

CREATE TABLE "tbl_company" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"company_name" NVARCHAR2(50), 
"city_id" NUMBER(19,0), 
"company_number" NVARCHAR2(50), 
"company_addr" NVARCHAR2(100), 
"company_addr_gis" NVARCHAR2(100), 
"company_tel" NVARCHAR2(20), 
"is_sub_company" NUMBER(19,0), 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COMPANY" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS";

CREATE TABLE "tbl_company_dot_map" (   
"id" NUMBER(19,0) NOT NULL ENABLE, 
"company_id" NUMBER(19,0), 
"dot_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COMPANY_DOT_MAP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_company_transport_route_ma" 
(
"id" NUMBER(19,0) NOT NULL ENABLE, 
"company_id" NUMBER(19,0), 
"transport_route_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COMPANY_TRANSPORT_ROUTE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_company_warehouse_map" ( 
"id" NUMBER(19,0) NOT NULL ENABLE, 
"company_id" NUMBER(19,0), 
"warehouse_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COMPANY_WAREHOUSE_MAP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_courier" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"job_num" NVARCHAR2(50), 
"name" NVARCHAR2(50), 
"birathday" DATE, 
"tel" NVARCHAR2(20), 
"pda_num" NVARCHAR2(50), 
"car_id" NUMBER(19,0), 
"postal_standard_id" NUMBER(19,0), 
"work_time_id" NUMBER(19,0), 
"dot_id" NUMBER(19,0), 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COURIER" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_deliver_region" 
(
"id" NUMBER(19,0) NOT NULL ENABLE, 
"search_keyword" NVARCHAR2(100), 
"search_assist_keyword" NVARCHAR2(100), 
"area_id" NUMBER(19,0), 
"fixed_area_id" NUMBER(19,0), 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DELIVER_REGION" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_delivery_record" (   
"id" NUMBER(19,0) NOT NULL ENABLE, 
"cur_warehouse_id" NVARCHAR2(50), 
"vehicle_id" NUMBER(19,0), 
"start_vehicle_dt" DATE,
"next_warehouse_id" NUMBER(19,0), 
"predict_arrivals_dt" DATE,
"actua_arrivals_dt" DATE,
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DELIVERY_RECORD" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_department" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"dep_name" NVARCHAR2(50), 
"dep_level" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DEPARTMENT" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_fixed_area" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"emp_id" NUMBER(19,0), 
"operator_dt" DATE, 
"operator_id" NUMBER(19,0), 
"gis_fence" NVARCHAR2(200), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_FIXED_AREA" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_goods_rack" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"warehouse_name" NVARCHAR2(50), 
"warehouse_addr" NVARCHAR2(100),
"warehouse_addr_gis" NVARCHAR2(50), 
"company_id" NUMBER(19,0), 
"employee_id" NVARCHAR2(200), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_GOODS_RACK" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_job" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"job_name" NVARCHAR2(50), 
"job_level" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_JOB" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_out_warehouse_detail" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"push_warehouse_id" NUMBER(19,0), 
"push_warehouse_bill" NVARCHAR2(100), 
"warehouse_id" NUMBER(19,0), 
"waybill_id" NUMBER(19,0), 
"pkg_id" NUMBER(19,0), 
"pkg_desc" NVARCHAR2(100), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_OUT_WAREHOUSE_DETAIL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_pkg" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"pw_bill" NVARCHAR2(50), 
"pw_dot_id" NUMBER(19,0), 
"warehouse_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_PKG" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_postal_standard" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"min_weight" NVARCHAR2(50), 
"min_length" NVARCHAR2(50), 
"max_length" NVARCHAR2(50), 
"trajectory" NVARCHAR2(50), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_POSTAL_STANDARD" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_push_warehouse_detail" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"push_warehouse_id" NUMBER(19,0), 
"push_warehouse_bill" NVARCHAR2(50), 
"warehouse_id" NUMBER(19,0), 
"pw_start_dt" NVARCHAR2(50), 
"pw_end_dt" NVARCHAR2(50), 
"pack_id" NUMBER(19,0), 
"pack_desc" NVARCHAR2(50), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_PUSH_WAREHOUSE_DETAIL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_service_evaluation" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"express_bill_id" NVARCHAR2(100), 
"express_bill" NUMBER(19,0), 
"pack_score" NUMBER(10,0), 
"delivery_time_score" NUMBER(10,0), 
"courier_score" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_SERVICE_EVALUATION" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_store_grid" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"warehouse_name" NVARCHAR2(50), 
"warehouse_addr" NVARCHAR2(100),
"warehouse_addr_gis" NVARCHAR2(50), 
"company_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_STORE_GRID" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_vehicle_monitor" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"delivery_record" NUMBER(19,0), 
"empId" NUMBER(19,0), 
"express_bill__id" NVARCHAR2(200), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_VEHICLE_MONITOR" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_rack_map" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"warehouse_name" NVARCHAR2(50), 
"warehouse_addr" NVARCHAR2(100),
"warehouse_addr_gis" NVARCHAR2(50), 
"company_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAREHOUSE_RACK_MAP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_receipt_detail" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"waybill_id" NUMBER(19,0), 
"pkg_id" NUMBER(19,0), 
"receipt_bill_id" NUMBER(19,0), 
"receipt_bill" NVARCHAR2(100), 
"operator_id" NUMBER(19,0), 
"state" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAREHOUSE_RECEIPT_DETAI" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_waybill_line" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"waybill_number" NVARCHAR2(100), 
"route_id" NUMBER(19,0), 
"serial_number" NVARCHAR2(100), 
"transport_tool" NUMBER(19,0), 
"delivery_record_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAYBILL_LINE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_waybill_state_record" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"waybill_id" NUMBER(19,0), 
"waybill_number" NVARCHAR2(100), 
"employee_id" NVARCHAR2(100), 
"consignee_id" NUMBER(19,0), 
"cur_warehouse_id" NUMBER(10,0), 
"next_warehouse_id" NUMBER(10,0), 
"deliverer_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAYBILL_STATE_RECORD" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_work_time" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"start_dt" NVARCHAR2(100),
"stop_dt" NVARCHAR2(100),
"saturday_start_dt" NVARCHAR2(100),
"saturday_stop_dt" NVARCHAR2(100),
"sunday_start_dt" NVARCHAR2(100),
"sunday_stop_dt" NVARCHAR2(100),
"state" NUMBER(10,0), 
"company_id" NUMBER(10,0), 
"operator_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WORK_TIME" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_test" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
 CONSTRAINT "PK_TBL_TEST" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_areas" (
"id" NUMBER(11,0) NOT NULL ENABLE, 
"name" NVARCHAR2(40), 
"pid" NUMBER(11,0), 
"sname" NVARCHAR2(40), 
"level" NVARCHAR2(11), 
"citycode" NVARCHAR2(20), 
"yzcode" NVARCHAR2(20), 
"mername" NVARCHAR2(100), 
"lng" NUMBER(11,4), 
"lat" NUMBER(11,4), 
"pinyin" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_AREAS" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_deliver_package" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"emp_id" NUMBER(19,0), 
"waybill_id" NUMBER(19,0), 
"waybill_number" NVARCHAR2(100), 
"express_bill_id" NUMBER(19,0), 
"express_bill_number" NVARCHAR2(100), 
"package_id" NUMBER(19,0), 
"collect_package_dt" DATE, 
"rece_type" NUMBER(19,0), 
"rece_dt" DATE, 
"state" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DELIVER_PACKAGE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_customer" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"tel" NVARCHAR2(20), 
"mobile" NVARCHAR2(20), 
"email" NVARCHAR2(50), 
"type" NUMBER(10,0), 
"is_own_reg" NUMBER(10,0), 
"reg_dt" DATE, 
"reg_channel_id" NUMBER(10,0), 
"state" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"last_login_dt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_CUSTOMER" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_codes" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"type" NUMBER(19,0), 
"code" NVARCHAR2(50), 
"code_desc" NVARCHAR2(100), 
"code_type" NVARCHAR2(50), 
"state" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
 CONSTRAINT "PK_TBL_CODES" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"addr" NVARCHAR2(19), 
"addr_gis" NVARCHAR2(50), 
"company_id" NUMBER(19,0), 
"employee_id" NUMBER(19,0), 
"type" NUMBER(10,0), 
"area" NVARCHAR2(50), 
"is_lease" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAREHOUSE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_consumer_address_map" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"consumer_id" NUMBER(19,0), 
"address_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_CUSTOMER_SENDER_MAP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_receipt" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"bill" NVARCHAR2(100), 
"type" NUMBER(19,0), 
"warehouse_id" NUMBER(19,0), 
"operator_id" NUMBER(19,0), 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAREHOUSE_RECEIPT" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_send_vehicle" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"out_warehouse_id" NUMBER(19,0), 
"out_warehouse_waybill_id" NUMBER(19,0), 
"out_warehouse_waybill_number" NVARCHAR2(100), 
"vehicle_id" NUMBER(19,0), 
"driver1_id" NUMBER(19,0), 
"driver2_id" NUMBER(19,0), 
"start_vehicle_dt" DATE, 
"next_warehouse_id" NUMBER(19,0), 
"predict_arrivals_dt" DATE, 
"actual_arrivals_dt" DATE, 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAREHOUSE_SEND_VEHICLE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_vehicle_map" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"warehouse_id" NUMBER(19,0), 
"vehicle_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COMPANY_VEHICLE_MAP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_dot" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"dot_number" NVARCHAR2(50), 
"dot_name" NVARCHAR2(50), 
"dot_addr" NVARCHAR2(100), 
"dot_gis_addr" NVARCHAR2(100), 
"dot_tel" NVARCHAR2(20), 
"company_id" NUMBER(19,0), 
"manage_area_id" NUMBER(19,0), 
"manage_area_gis" NVARCHAR2(100), 
"state" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DOT" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_transport_tool" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"brand" NVARCHAR2(100), 
"model" NVARCHAR2(100), 
"type" NUMBER(19,0), 
"given_load" NVARCHAR2(100), 
"load_cn_unit" NVARCHAR2(100), 
"load_en_unit" NVARCHAR2(100), 
"buy_dt" DATE, 
"license_plate" NVARCHAR2(100), 
"state" NVARCHAR2(100), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_TRANSPORT_TOOL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_dot_transport_tool" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"dot_id" NUMBER(19,0), 
"transport_tool_id" NUMBER(19,0), 
"allocate_dt" DATE, 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_DOT_TRANSPORT_TOOL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_address" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"name" NVARCHAR2(50), 
"tel" NVARCHAR2(20), 
"mobile" NVARCHAR2(20), 
"detail_addr" NVARCHAR2(100), 
"area_id" NUMBER(19,0), 
"gis_addr" NVARCHAR2(20), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_CUSTOMER_ADDRESS" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_route" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"start_station" NVARCHAR2(50), 
"start_station_area_id" NUMBER(19,0), 
"start_warehouse_id" NUMBER(19,0), 
"end_station" NVARCHAR2(50), 
"end_station_area_id" NUMBER(19,0), 
"end_warehouse_id" NUMBER(19,0), 
"mileage_m" NUMBER(10,0), 
"time_consumer_minute" NUMBER(10,0), 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_ROUTE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_push_warehouse" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"pw_waybill_id" NUMBER(19,0), 
"pw_waybill_number" NVARCHAR2(50), 
"pw_dot_id" NUMBER(19,0), 
"warehouse_id" NUMBER(19,0), 
"emp_id" NUMBER(19,0), 
"pw_start_dt" DATE, 
"pw_end_dt" DATE, 
"pw_position" NVARCHAR2(50), 
"pw_reg_emp_id" NUMBER(19,0), 
"ow_reg_emp_scan_gun_id" NUMBER(19,0), 
"pw_confirm_emp_id" NUMBER(19,0), 
"ow_confirm_emp_scan_gun_id" NUMBER(19,0), 
"pw_box_emp_id" NUMBER(19,0), 
"pw_box_scan_gun_id" NUMBER(19,0), 
"pw_after_seal_img" NVARCHAR2(100), 
"pw_receipt_number" NVARCHAR2(100), 
"pw_receipt_dt" DATE, 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_PUSH_WAREHOUSE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_out_warehouse" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"pw_waybill_id" NUMBER(19,0), 
"pw_waybill_number" NVARCHAR2(100), 
"ow_dot_id" NUMBER(19,0), 
"warehouse_id" NUMBER(19,0), 
"ow_vehicle_id" NUMBER(19,0), 
"ow_driver_emp_id" NUMBER(19,0), 
"ow_follow1_emp_id" NUMBER(19,0), 
"ow_follow2_emp_id" NUMBER(19,0), 
"ow_start_dt" DATE, 
"ow_end_dt" DATE, 
"ow_position" NVARCHAR2(50), 
"ow_reg_emp_id" NUMBER(19,0), 
"ow_reg_scan_gun_id" NUMBER(19,0), 
"ow_confirm_emp_id" NUMBER(19,0), 
"ow_confirm_scan_gun_id" NUMBER(19,0), 
"ow_pre_seal_img" NVARCHAR2(100), 
"ow_receipt_number" NVARCHAR2(100), 
"ow_receipt_dt" DATE, 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_OUT_WAREHOUSE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_warehouse_emp" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"job_num" NVARCHAR2(50), 
"name" NVARCHAR2(50), 
"birathday" DATE, 
"tel" NVARCHAR2(20), 
"type" NUMBER(10,0), 
"warehouse_id" NUMBER(19,0), 
"state" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAREHOUSE_EMP" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_express_package" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"scan_gun_id" NVARCHAR2(19), 
"name" NVARCHAR2(50), 
"cid" NUMBER(10,2), 
"weight" NUMBER(10,2), 
"amount" NUMBER(10,2), 
"coupon_id" NUMBER(19,0), 
"coupon_amount" NUMBER(10,2), 
"actual_amount" NUMBER(10,2), 
"insured_price" NUMBER(10,2), 
"is_fragile" NVARCHAR2(20), 
"send_address_id" NUMBER(19,0), 
"recv_address_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_EXPRESS_PACKAGE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_express_bill" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"express_number" NVARCHAR2(50), 
"cid" NUMBER(19,0), 
"eid" NUMBER(19,0), 
"order_channel_id" NUMBER(19,0), 
"order_dt" DATE, 
"order_terminal_type" NUMBER(10,0), 
"order_terminal_os_type" NUMBER(10,0), 
"reserve_dt" DATE, 
"is_collect_package_timeout" NUMBER(10,0), 
"timeout_dt" DATE, 
"type" NUMBER(10,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_EXPRESS_BILL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_consumer_sender_info" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"ciid" NUMBER(19,0), 
"pkg_id" NUMBER(19,0), 
"express_bill_id" NUMBER(19,0), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_CUSTOMER_SENDER_INFO" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_collect_package" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"cid" NUMBER(19,0), 
"eid" NUMBER(19,0), 
"pkg_id" NUMBER(19,0), 
"express_bill_id" NUMBER(19,0), 
"express_bill_number" NVARCHAR2(100), 
"state" NUMBER(10,0), 
"collect_package_dt" DATE, 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_COLLECT_PACKAGE" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS"; 

CREATE TABLE "tbl_waybill" (
"id" NUMBER(19,0) NOT NULL ENABLE, 
"express_bill_number" NVARCHAR2(100), 
"waybill_number" NVARCHAR2(100), 
"cid" NUMBER(19,0), 
"eid" NUMBER(19,0), 
"order_channel_id" NUMBER(19,0), 
"order_dt" DATE, 
"order_terminal_type" NUMBER(10,0), 
"order_terminal_os_type" NUMBER(10,0), 
"reserve_dt" DATE, 
"is_collect_package_timeout" NUMBER(10,0), 
"pkg_id" NUMBER(19,0), 
"pkg_number" NVARCHAR2(100), 
"timeout_dt" NVARCHAR2(100), 
"transform_type" NUMBER(10,0), 
"delivery_customer_name" NVARCHAR2(100), 
"delivery_addr" NVARCHAR2(100), 
"delivery_mobile" NVARCHAR2(100), 
"delivery_tel" NVARCHAR2(100), 
"receive_customer_name" NVARCHAR2(100), 
"receive_addr" NVARCHAR2(100), 
"receive_mobile" NVARCHAR2(100), 
"receive_tel" NVARCHAR2(100), 
"cdt" DATE, 
"udt" DATE, 
"remark" NVARCHAR2(100), 
 CONSTRAINT "PK_TBL_WAYBILL" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS";

-- 运输记录表
CREATE TABLE "tbl_transport_record" (  
   "id" NUMBER(19,0) NOT NULL ENABLE, 
   "pw_id" NUMBER(19,0), 
   "pw_waybill_id" NUMBER(19,0), 
   "pw_waybill_number" NVARCHAR2(100), 
   "ow_id" NUMBER(19,0), 
   "ow_waybill_id" NUMBER(19,0), 
   "ow_waybill_number" NVARCHAR2(100), 
   "sw_id" NUMBER(19,0), 
   "ew_id" NUMBER(19,0), 
   "transport_tool_id" NUMBER(19,0), 
   "pw_driver1_id" NUMBER(19,0), 
   "pw_driver2_id" NUMBER(19,0), 
   "pw_driver3_id" NUMBER(19,0), 
   "ow_driver1_id" NUMBER(19,0), 
   "ow_driver2_id" NUMBER(19,0), 
   "ow_driver3_id" NUMBER(19,0), 
   "route_id" NUMBER(19,0), 
   "distance" NUMBER(10,0), 
   "duration" NUMBER(10,0), 
   "state" NUMBER(10,0), 
   "start_vehicle_dt" DATE, 
   "predict_arrivals_dt" DATE, 
   "actual_arrivals_dt" DATE, 
   "cdt" DATE, 
   "udt" DATE, 
   "remark" NVARCHAR2(100), 
    CONSTRAINT "PK_TBL_TRANSPORT_RECORD" PRIMARY KEY ("id")
) TABLESPACE "TBS_LOGISTICS";

2、配置管理器MGR进程

  • 进入源端OGG命令行

./ggsci

创建mgr进程：edit param mgr

PORT 7809 DYNAMICPORTLIST 7810-7909 AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3 PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

3、​​​​​​​配置extract进程

配置Extract进程：edit param extkafka

新增内容：

extract extkafka GETUPDATEBEFORES NOCOMPRESSDELETES NOCOMPRESSUPDATES dynamicresolution SETENV (ORACLE_SID = "orcl") SETENV (NLS_LANG = "american_america.AL32UTF8") userid ogg,password ogg exttrail /u01/app/ogg/src/dirdat/to table ogg.*;

参数名称

参数说明

extract extkafka

定义extract进程名称

dynamicresolution

启用动态解析

SETENV (ORACLE_SID = "orcl")

设置Oracle数据库

SETENV (NLS_LANG = "american_america.AL32UTF8")

设置字符集

userid ogg,password ogg

ogg连接Oracle数据库的帐号密码

exttrail /u01/app/ogg/src/dirdat/to

定义trail文件的保存位置以及文件名，文件字母最多2个，否则会报错

table ogg.*;

复制表的表名，支持*通配，必须以;结尾

4、配置pump进程

配置Pump进程：edit param pukafka

新增内容：

extract pukafka passthru dynamicresolution userid ogg,password ogg rmthost localhost mgrport 7809 rmttrail /u01/app/ogg/tgr/dirdat/to table ogg.*;

extract进程名称；passthru即禁止OGG与Oracle交互，我们这里使用pump逻辑传输，故禁止即可；dynamicresolution动态解析；userid ogg,password ogg即OGG连接Oracle数据库的帐号密码rmthost和mgrhost即目标端(kafka)OGG的mgr服务的地址以及监听端口；rmttrail即目标端trail文件存储位置以及名称。

参数名称

参数说明

extract pukafka

定义pump进程名称

passthru

因使用了pump逻辑传输，所以禁止OGG与Oracle交互

dynamicresolution

配置动态解析

userid ogg,password ogg

OGG连接Oracle数据库的帐号密码

rmthost localhost mgrport 7809

目标端OGG的mgr服务的地址以及监听端口

rmttrail /u01/app/ogg/tgr/dirdat/to

目标端OGG的trail文件存储位置以及名称

table ogg.*;

要采集的表,必须使用;结尾

5、​​​​​​​配置define文件

注意：该文件用来在异构数据源之间传输时，需明确知道表之间的映射关系，比如： Oracle与MySQL，Hadoop集群（HDFS，Hive，kafka等）等之间数据传输可以定义为异构数据类型的传输，故需要定义表之间的关系映射，在OGG命令行执行：

配置define文件：edit param test_ogg

defsfile /u01/app/ogg/src/dirdef/test_ogg.test_ogg userid ogg,password ogg table ogg.*;

生成表schema文件：（在OGG_SRC_HOME目录下执行(oracle用户)）

./defgen paramfile dirprm/test_ogg.prm

将生成的/u01/app/ogg/src/dirdef/test_ogg.test_ogg发送的目标端ogg目录下的dirdef里：

scp -r /u01/app/ogg/src/dirdef/test_ogg.test_ogg   /u01/app/ogg/tgr/dirdef/

因为目标端目录还没有创建，因此发送文件可能会失败，所以执行完目标端配置后发送即可

三、配置OGG目标端

1、​​​​​​​拷贝源端的define文件到目标端

将生成的/u01/app/ogg/src/dirdef/test_ogg.test_ogg发送的目标端ogg目录下的dirdef里：

scp -r $OGG_SRC_HOME/dirdef/test_ogg.test_ogg $OGG_TGR_HOME/dirdef/

2、​​​​​​​配置管理器MRG进程

操作步骤

说明

1

使用oracle用户进入OGG_SRC_HOME目录下

切换到oracle用户下：su – oracle

打印目标端OGG_TGR_HOME：echo $OGG_TGR_HOME

进入OGG_TGR_HOME：cd $OGG_TGR_HOME

启动ggsci：./ggsci

2

配置目标端MRG进程

配置MGR进程：edit param mgr

新增内容：

PORT 7810 DYNAMICPORTLIST 7810-7909 AUTORESTART EXTRACT *,RETRIES 5,WAITMINUTES 3 PURGEOLDEXTRACTS ./dirdat/*,usecheckpoints, minkeepdays 3

3、配置Replicate进程

操作步骤

说明

1

配置目标端Replicate进程

配置replicate进程：edit param rekafka

REPLICAT rekafka sourcedefs /u01/app/ogg/tgr/dirdef/test_ogg.test_ogg TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props REPORTCOUNT EVERY 1 MINUTES, RATE GROUPTRANSOPS 10000 MAP ogg.*, TARGET ogg.*;

4、​​​​​​​配置kafka.props

  • 配置kafka.props

cd $OGG_TGR_HOME

vim dirprm/kafka.props

新增内容：

gg.handlerlist=kafkahandler gg.handler.kafkahandler.type=kafka gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties gg.handler.kafkahandler.topicMappingTemplate=logistics gg.handler.kafkahandler.format=json gg.handler.kafkahandler.mode=op gg.classpath=dirprm/:/export/services/kafka/libs/*:/u01/app/ogg/tgr/:/u01/app/ogg/tgr/lib/*

  • 配置custom_kafka_producer.properties

cd $OGG_TGR_HOME

vim dirprm/custom_kafka_producer.properties

新增内容：

bootstrap.servers=node2:9092 acks=1 compression.type=gzip reconnect.backoff.ms=1000 value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer batch.size=102400 linger.ms=10000

​​​​​​​5、最后确认所有的进程

在目标端，主要做了4个操作，共包括2个进程，分别是MANAGER和REPLICAT。

四、​​​​​​​OGG测试

1、​​​​​​​启动kafka消费者

操作步骤

说明

1

启动node2服务器（大数据服务器）

2

启动kafka消费者

kafka-console-consumer --bootstrap-server node2:9092 --topic logistics

3

启动oracle客户端

插入一条数据

INSERT INTO OGG."tbl_company"("id", "company_name", "city_id", "company_number", "company_addr", "company_addr_gis", "company_tel", "is_sub_company", "state", "cdt", "udt", "remark") VALUES(11, '广州速递邮箱公司', 440100, NULL, '广州', '117.28177895734918_31.842711680531399', NULL, 1, 1, TO_DATE('2020-06-13 15:24:51','yyyy-mm-dd hh24:mi:ss'), TO_DATE('2020-06-13 15:24:51','yyyy-mm-dd hh24:mi:ss'), NULL);

4

查看kafka消费者是否可以打印出来日志

修改一条数据

UPDATE OGG."tbl_company" SET "company_name"='广州速递有限公司-1' WHERE "id"=11;

2、​​​​​​​archivelog日志路径

kafka oracle r语言

0 人点赞