使用Embulk从Oracle抽取数据到Trafodion

在前面的文章&＃xff0c;我们介绍过使用开源工具Kettle可以实现关系型数据库如Oracle到Trafodion的数据抽取工作&＃xff0c;本文介绍另外一种ETL工具Embulk。Embulk是一款开源的批量加载工具&＃xff0c;可以在多种数据库、存储引擎、文本格式及云服务之间进行数据转换&＃xff0c;Embulk官网中的下面这张图比较简单明了的描述出了其特点&＃xff0c;

Embulk提供很多plugins&＃xff0c;包括Input plugins、Output plugins、File parser plugins等&＃xff0c;详细可以参考http://www.embulk.org/plugins/。用户也可以开发自己的plugins并上传&＃xff0c;如EsgynDB就基于trafodion自己开发一款基于trafodion的upsert using load实现的批量加载插件&＃xff0c;如下图&＃xff0c;

本文在此利用上述的trafodion output插件及oracle input插件通过实例描述如何使用embulk实现oracle到trafodion的数据加载。
1 下载安装Embulk

wget https://dl.embulk.org/embulk-latest.jar
mkdir ~/.embulk/bin
mv embulk-latest.jar ~/.embulk/bin/embulk
chmod &＃43;x ~/.embulk/bin/embulk
echo &＃39;export PATH&＃61;"$HOME/.embulk/bin:$PATH"&＃39; >> ~/.bashrc
source ~/.bashrc

2 验证Embulk下载安装成功

[root&＃64;n12 ~]# embulk gem list
2017-09-05 20:19:51.184 &＃43;0800: Embulk v0.8.31*** LOCAL GEMS ***did_you_mean (default: 1.0.1)
jar-dependencies (default: 0.3.5)
jruby-openssl (0.9.17 java)
json (1.8.3 java)
minitest (default: 5.4.1)
net-telnet (default: 0.1.1)
power_assert (default: 0.2.3)
psych (2.0.17 java)
racc (1.4.14 java)
rake (default: 10.4.2)
rdoc (default: 4.2.0)
test-unit (default: 3.1.1)

3 安装oracle input插件及trafodion output插件

embulk gem install embulk-input-oracle
embulk gem install embulk-output-trafodion

4 验证以上插件安装成功

root&＃64;n12 ~]# embulk gem list
2017-09-05 20:22:57.146 &＃43;0800: Embulk v0.8.31*** LOCAL GEMS ***did_you_mean (default: 1.0.1)
embulk-input-oracle (0.8.5)
embulk-output-trafodion (0.1.1)
jar-dependencies (default: 0.3.5)
jruby-openssl (0.9.17 java)
json (1.8.3 java)
minitest (default: 5.4.1)
net-telnet (default: 0.1.1)
power_assert (default: 0.2.3)
psych (2.0.17 java)
racc (1.4.14 java)
rake (default: 10.4.2)
rdoc (default: 4.2.0)
test-unit (default: 3.1.1)

5 准备oracle jar包&＃xff0c;用于读取oracle数据库

[root&＃64;n12 ~]# ll /opt/drivers/
total 2676
-rw-rw-r-- 1 trafodion trafodion 2739670 Sep 5 20:38 ojdbc6.jar

6 编辑YAML文件oracle_to_trafodion.yml
注&＃xff1a;Embulk使用YAML文件来定义数据批量加载的方式&＃xff0c;YAML文件格式定义可参考 http://www.embulk.org/docs/built-in.html#embulk-configuration-file-format

exec:max_threads: 8min_output_tasks: 4
in:type: oracledriver_path: /opt/drivers/ojdbc6.jarurl: "jdbc:oracle:thin:&＃64;10.10.11.16:1521:esgyn"user: systempassword: system12345query: |select * from test_tbl
out:type: trafodionurl: "jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase"user: trafodionpassword: traf123default_timezone: &＃39;Asia/Shanghai&＃39;table: TEST_TBLmode: insert_direct

7 创建Oracle测试表并插入测试数据

SQL> create table test_tbl(a char(10),b char(10));
SQL> insert into test_tbl values(&＃39;a&＃39;,&＃39;b&＃39;);
SQL> insert into test_tbl values(&＃39;c&＃39;,&＃39;d&＃39;);
SQL> insert into test_tbl values(&＃39;e&＃39;,&＃39;f&＃39;);
SQL> select * from test_tbl;A B
---------- ----------
a b
c d
e f

8 创建Trafodion目标表

>>create table test_tbl(a char(10),b char(10));--- SQL operation complete.
>>select * from test_tbl;--- 0 row(s) selected.

9 运行Embulk任务

--预览
embulk preview ~/oracle_to_trafodion.yml
[root&＃64;n12 ~]# embulk preview ~/oracle_to_trafodion.yml
2017-09-05 20:41:24.093 &＃43;0800: Embulk v0.8.31
2017-09-05 20:41:27.303 &＃43;0800 [INFO] (0001:preview): Loaded plugin embulk-input-oracle (0.8.5)
2017-09-05 20:41:27.662 &＃43;0800 [INFO] (0001:preview): Using JDBC Driver 11.2.0.4.0
2017-09-05 20:41:27.977 &＃43;0800 [INFO] (0001:preview): SQL: select * from test_tbl2017-09-05 20:41:27.999 &＃43;0800 [INFO] (0001:preview): > 0.02 seconds
&＃43;------------&＃43;------------&＃43;
| A:string | B:string |
&＃43;------------&＃43;------------&＃43;
| a | b |
| c | d |
| e | f |
&＃43;------------&＃43;------------&＃43;--运行
embulk run ~/oracle_to_trafodion.yml
[root&＃64;n12 ~]# embulk run ~/oracle_to_trafodion.yml
2017-09-05 21:01:26.419 &＃43;0800: Embulk v0.8.31
2017-09-05 21:01:30.260 &＃43;0800 [INFO] (0001:transaction): Loaded plugin embulk-input-oracle (0.8.5)
2017-09-05 21:01:30.298 &＃43;0800 [INFO] (0001:transaction): Loaded plugin embulk-output-trafodion (0.1.1)
2017-09-05 21:01:30.807 &＃43;0800 [INFO] (0001:transaction): Using JDBC Driver 11.2.0.4.0
2017-09-05 21:01:30.995 &＃43;0800 [INFO] (0001:transaction): Using local thread executor with max_threads&＃61;8 / output tasks 4 &＃61; input tasks 1 * 4
-----------------------------------jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase
2017-09-05 21:01:31.019 &＃43;0800 [INFO] (0001:transaction): Connecting to jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase options {user&＃61;trafodion, tcpKeepAlive&＃61;true, useCompression&＃61;true, rewriteBatchedStatements&＃61;true, connectTimeout&＃61;300000, socketTimeout&＃61;1800000}
connection------------------------------------:org.trafodion.jdbc.t4.TrafT4Connection&＃64;1cc41b77
2017-09-05 21:01:36.271 &＃43;0800 [INFO] (0001:transaction): Using insert_direct mode
[WARN] Plugin uses deprecated constructor of org.embulk.spi.time.TimestampFormatter.
[WARN] Report plugins in your config at: https://github.com/embulk/embulk/issues/745
2017-09-05 21:01:40.345 &＃43;0800 [INFO] (0001:transaction): {done: 0 / 1, running: 0}
-----------------------------------jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase
2017-09-05 21:01:40.389 &＃43;0800 [INFO] (0026:task-0000): Connecting to jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase options {user&＃61;trafodion, tcpKeepAlive&＃61;true, useCompression&＃61;true, rewriteBatchedStatements&＃61;true, connectTimeout&＃61;300000, socketTimeout&＃61;1800000}
connection------------------------------------:org.trafodion.jdbc.t4.TrafT4Connection&＃64;4ee9df38
2017-09-05 21:01:40.970 &＃43;0800 [INFO] (0026:task-0000): Prepared SQL: UPSERT USING LOAD INTO "TEST_TBL" ("A", "B") VALUES (?, ?)
-----------------------------------jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase
2017-09-05 21:01:47.001 &＃43;0800 [INFO] (0026:task-0000): Connecting to jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase options {user&＃61;trafodion, tcpKeepAlive&＃61;true, useCompression&＃61;true, rewriteBatchedStatements&＃61;true, connectTimeout&＃61;300000, socketTimeout&＃61;1800000}
connection------------------------------------:org.trafodion.jdbc.t4.TrafT4Connection&＃64;785aae0e
2017-09-05 21:01:52.935 &＃43;0800 [INFO] (0026:task-0000): Prepared SQL: UPSERT USING LOAD INTO "TEST_TBL" ("A", "B") VALUES (?, ?)
-----------------------------------jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase
2017-09-05 21:02:01.036 &＃43;0800 [INFO] (0026:task-0000): Connecting to jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase options {user&＃61;trafodion, tcpKeepAlive&＃61;true, useCompression&＃61;true, rewriteBatchedStatements&＃61;true, connectTimeout&＃61;300000, socketTimeout&＃61;1800000}
connection------------------------------------:org.trafodion.jdbc.t4.TrafT4Connection&＃64;38f4c238
2017-09-05 21:02:03.872 &＃43;0800 [INFO] (0026:task-0000): Prepared SQL: UPSERT USING LOAD INTO "TEST_TBL" ("A", "B") VALUES (?, ?)
-----------------------------------jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase
2017-09-05 21:02:10.364 &＃43;0800 [INFO] (0026:task-0000): Connecting to jdbc:t4jdbc://192.168.1.93:23400/:schema&＃61;seabase options {user&＃61;trafodion, tcpKeepAlive&＃61;true, useCompression&＃61;true, rewriteBatchedStatements&＃61;true, connectTimeout&＃61;300000, socketTimeout&＃61;1800000}
connection------------------------------------:org.trafodion.jdbc.t4.TrafT4Connection&＃64;6c47ecb4
2017-09-05 21:02:17.565 &＃43;0800 [INFO] (0026:task-0000): Prepared SQL: UPSERT USING LOAD INTO "TEST_TBL" ("A", "B") VALUES (?, ?)
2017-09-05 21:02:36.756 &＃43;0800 [INFO] (0026:task-0000): SQL: select * from test_tbl2017-09-05 21:02:36.777 &＃43;0800 [INFO] (0026:task-0000): > 0.02 seconds
2017-09-05 21:02:36.783 &＃43;0800 [INFO] (0026:task-0000): Loading 3 rows
2017-09-05 21:02:36.821 &＃43;0800 [INFO] (0026:task-0000): > 0.04 seconds (loaded 3 rows in total)
2017-09-05 21:02:37.653 &＃43;0800 [INFO] (0001:transaction): {done: 1 / 1, running: 0}
2017-09-05 21:02:37.665 &＃43;0800 [INFO] (main): Committed.
2017-09-05 21:02:37.665 &＃43;0800 [INFO] (main): Next config diff: {"in":{},"out":{}}

10 检查Embulk任务执行成功

SQL>select * from test_tbl;A B
---------- ----------
a b
c d
e f--- 3 row(s) selected.

至此&＃xff0c;使用Embulk从Oracle抽取数据到Trafodion演示完毕&＃xff01;