1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
// MysqlUtil.java
package com.example;
import java.sql.*;
public class MysqlUtil {
private Connection conn = null;
private Statement stmt = null;
// vm_mysql_2
private static final String DB_URL = "jdbc:mysql://192.168.136.129:3306/mydatabase?characterEncoding=utf-8";
private static final String DB_DRIVER = "com.mysql.cj.jdbc.Driver";
private static final String DB_USERNAME = "root";
private static final String DB_PASSWORD = "root";
public Connection getConnection() throws ClassNotFoundException {
try{
Class.forName(DB_DRIVER);
conn = DriverManager.getConnection(DB_URL,DB_USERNAME,DB_PASSWORD);
stmt = conn.createStatement();
} catch(Exception e){
e.printStackTrace();
} finally {
return conn;
}
}
public ResultSet executeQuery(String sql){
ResultSet rs = null;
if(sql != null && sql.length() != 0){
try{
this.getConnection();
//this.stmt.execute(sql);
rs = stmt.executeQuery(sql);
return rs;
} catch(SQLException e){
e.printStackTrace();
} catch (ClassNotFoundException e) {
e.printStackTrace();
} finally{
if(this.stmt != null){
//this.stmt.close();
}
}
return rs;
}else{
return rs;
}
}
public boolean executeUpdate(String sql) throws SQLException {
if(sql != null && sql.length() != 0){
try{
this.getConnection();
this.stmt.executeUpdate(sql);
return true;
} catch(SQLException e){
e.printStackTrace();
} catch (ClassNotFoundException e) {
e.printStackTrace();
} finally{
if(this.stmt != null){
this.stmt.close();
}
}
return false;
}else{
return false;
}
}
public static void main(String[] args) throws SQLException {
MysqlUtil mysqlUtil = new MysqlUtil();
String sql = "select * from testTable";
ResultSet rs = mysqlUtil.executeQuery(sql);
while(rs.next()){
System.out.println(rs.getString("sname")+
rs.getString("sno")+
rs.getString("classno"));
}
rs.close();
mysqlUtil.stmt.close();
mysqlUtil.conn.close();
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
//ExcelLoad.java
package com.example;
import jxl.Cell;
import jxl.Sheet;
import jxl.Workbook;
import jxl.read.biff.BiffException;
import java.io.File;
import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
public class ExcelLoad {
public static final String tableName = "testTable";
static final int sheetNum = 0;
static final int startColumn = 1;
static final int endColumn = 3;
static final int startRow = 2;
private int endRow ;
public void HandleSingleFile(File excelFile){
System.out.println("正在处理文件：" + excelFile.getName());
Workbook wb = null;
try{
wb = Workbook.getWorkbook(excelFile);
} catch (IOException e) {
e.printStackTrace();
} catch (BiffException e) {
e.printStackTrace();
}
//读入页
Sheet sheet = wb.getSheets()[sheetNum];
this.endRow = sheet.getRows();
for(int i = startRow - 1; i < this.endRow; i++){
List<String> rowData = new ArrayList<String>();
for(int j = startColumn -1; j < endColumn; j++){
Cell cell = sheet.getCell(j,i);
String cellContent = cell.getContents();
if(cellContent == ""){
cellContent = null;
}
rowData.add(cellContent);
}
System.out.println("正在处理第"+(i+1)+"行"+rowData.get(0));
try{
insertRow(rowData, tableName);
} catch (SQLException throwables) {
throwables.printStackTrace();
}
}
}
public void insertRow(List<String> rowData, String dataTable) throws SQLException {
String sql = "INSERT INTO testTable (sname,sno,classno) VALUES (";
sql += "'"+rowData.get(0).toString()+"'";
sql += ",";
sql += "'" + rowData.get(1).toString() + "'";
sql += ",";
sql += "'" + rowData.get(2).toString() + "'";
sql += ") ON DUPLICATE KEY UPDATE sname=values(sname),classno=values(classno)";
MysqlUtil mysqlUtil = new MysqlUtil();
mysqlUtil.executeUpdate(sql);
}
public static void main(String[] args){
String path = "D:\document\blog\kettle\introduce\student.xls";
File file = new File(path);
ExcelLoad excelLoad = new ExcelLoad();
excelLoad.HandleSingleFile(file);
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import pandas as pd
import pymysql as pm
import numpy as np
path = 'D:/document/blog/kettle/introduce/student.xls'
# vm_mysql_3
dbhost = '192.168.136.133'
dbport = 3306
dbname = 'mydatabase'
username = 'root'
pwd = 'root'
sql = '''INSERT INTO testTable (sname,sno,classno) VALUES (%s,%s,%s)
ON DUPLICATE KEY UPDATE sname=values(sname),classno=values(classno)
'''
def loadExcel():
global path
excelFile = pd.read_excel(path)
data_list = np.array(excelFile)
data_target_list=[]
for row_src in data_list:
row_target = (row_src[0],row_src[1],row_src[2])
data_target_list.append(row_target)
#
print(data_list)
print(data_target_list)
global sql
sqlExcute(sql,data_target_list)
def sqlExcute(sqlStr,datatuple):
global dbhost,dbport,dbname,username,password
conn = pm.connect(host=dbhost,user=username,password=pwd,db=dbname,port=dbport)
cur = conn.cursor()
try:
cur.executemany(sqlStr,datatuple)
conn.commit()
except Exception as e:
conn.rollback()
finally:
conn.close()
if __name__ == '__main__':
loadExcel()

1
2
3
4
time /t
D:pSoftkettledata-integrationpan.bat /file D:documentblogkettleintroduceexcelLoad.ktr /level Basic /logfile D:documentblogkettleintroducetest1.log
time /t

1
2
/usr/local/kettle/pan.sh -file /tmp/kettleFile/mysql_2to1_linux.ktr /level Basic /logfile /tmp/kettleFile/test1.log

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
package demo;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
public class Demo {
public static void main(String[] args) throws KettleException {
//初始化ketlle
KettleEnvironment.init();
//创建转换元数据对象
TransMeta meta = new TransMeta("D:\document\blog\kettle\introduce\mysql_2to1.ktr");
Trans trans = new Trans(meta);
trans.prepareExecution(null);
trans.startThreads();
trans.waitUntilFinished();
if(trans.getErrors()==0){
System.out.println("执行成功！");
}
}
}