create table poc_app.sys_ftp_cfg ( ftp_id varchar(100) not null, --话单文件名标记 ftp_cycle_id varchar(1) not null, --话单文件名周期 ftp_stage_filepath varchar(255) not null, --话单处理后路径 ftp_stage_filereg varchar(100) not null, --话单
create table poc_app.sys_ftp_cfg
(
ftp_id varchar(100) not null, --话单文件名标记
ftp_cycle_id varchar(1) not null, --话单文件名周期
ftp_stage_filepath varchar(255) not null, --话单处理后路径
ftp_stage_filereg varchar(100) not null, --话单处理后名称格式
stage_schema varchar(100) not null, --schema名称
table_name varchar(100) not null, --表名
delimiter_type_id varchar(10) not null --分隔符
);
insert into poc_app.sys_ftp_cfg
values('jiang_test_d','d','/home/sybase/day','jiang_test_[yyyymmdd].dat','poc_app','jiang_test','|');
#!/usr/bin/python
#-*- encoding: utf-8 -*-
####################################################################################
# name: sybaseiq_loaddata.py
# describe: 通过load table命令将数据文件加载到sybase iq数据库里面
####################################################################################
import os
import pyodbc
import string
import sys
from subprocess import popen,pipe
import configparser
reload(sys)
sys.setdefaultencoding('utf8')
'''
将数据文件加载到sybase iq数据库里面
'''
class sybaseiqload:
debug = 0
def __init__(self,dbinfo):
self.uid = dbinfo[1]
self.pwd = dbinfo[2]
odbcinfo = 'dsn=%s;uid=%s;pwd=%s'%(dbinfo[0],dbinfo[1],dbinfo[2])
self.cnxn = pyodbc.connect(odbcinfo,autocommit=true,ansi=true)
self.cursor = self.cnxn.cursor()
def __del__(self):
if self.cursor:
self.cursor.close()
if self.cnxn:
self.cnxn.close()
def _printinfo(self,msg):
print %s%(msg)
print \n
def _getstagename(self,ftp_stage_filereg,ftp_cycle_id,cur_static_time):
if ftp_cycle_id.lower() == 'h':
ftp_stage_filename = ftp_stage_filereg.replace('[yyyymmddhh]',cur_static_time[0:10])
if ftp_cycle_id.lower() == 'd':
ftp_stage_filename = ftp_stage_filereg.replace('[yyyymmdd]',cur_static_time[0:8])
if ftp_cycle_id.lower() == 'w':
ftp_stage_filename = ftp_stage_filereg.replace('[yyyy_ww]',cur_static_time[0:7])
if ftp_cycle_id.lower() == 'm':
ftp_stage_filename = ftp_stage_filereg.replace('[yyyymm]',cur_static_time[0:6])
return ftp_stage_filename
def _getloadinfo(self,ftp_id):
sql = '''
select
ftp_cycle_id
,ftp_stage_filepath
,ftp_stage_filereg
,stage_schema
,delimiter_type_id
,table_name
from jiang.sys_ftp_cfg
where ftp_id = '%s'
''' %(ftp_id)
self.cursor.execute(sql.strip())
row = self.cursor.fetchone()
return row
def _getsybiqservinfo(self):
# 保存sybaseiq的主机和端口号
sybservinfo = []
# odbc配置文件绝对路径
unixodbc_file = /etc/unixodbc/odbc.ini
config = configparser.configparser()
config.read(unixodbc_file)
# 获取sybaseiq的ip地址
serverip = config.get(sybaseiqdsn, server)
# 获取sybaseiq的端口号
port = config.get(sybaseiqdsn, port)
# 保存获取的ip地址和端口号
sybservinfo.append(serverip)
sybservinfo.append(port)
return sybservinfo
def loaddata(self,ftp_id,cur_static_time):
#取文件加载相关配置信息
row = self._getloadinfo(ftp_id)
ftp_cycle_id = row[0]
ftp_stage_filepath = row[1]
ftp_stage_filereg = row[2]
stage_schema = row[3]
delimiter_type_id = row[4]
table_name = row[5]
# 获取指定日期的文件名
ftp_stage_filename = self._getstagename(ftp_stage_filereg,ftp_cycle_id,cur_static_time)
# 获取清洗后文件的绝对路径
ftp_stage_absolute_filename = os.path.join(ftp_stage_filepath,ftp_stage_filename)
# 对清洗后的文件再进行处理
#ftp_stage_absolute_filename_final = ftp_stage_absolute_filename + '*'
# 获取sybaseiq的主机ip地址和端口号
sybaseiq_ipport = self._getsybiqservinfo()
# 获取表的所有字段
table_columns = '''
select column_name
from syscolumn a
join systable b
on a.table_id = b.table_id
where b.table_name = '%s' ># /tmp/table_name.log
'''%(table_name)
load_sql='''dbisql -c uid=%s;pwd=%s -host %s -port %s -nogui %s'''%(self.uid,self.pwd,sybaseiq_ipport[0],sybaseiq_ipport[1],table_columns)
os.system(load_sql)
# 处理生成的表字段文件
columns_sql = '''
cat /tmp/table_name.log | sed s/'//g | awk '{printf %s,,$0}'| sed 's/,$//g'
'''
result = popen(columns_sql,shell=true,stdout=pipe,stderr=pipe)
right_info = result.stdout.read().strip('\xef|\xbb|\xbf')
err_info = result.stderr.read()
loadsql = '''
load table %s.cpms_area_user
(
%s
)
using file '%s'
format ascii
escapes off
quotes off
notify 1000000
delimited by '%s'
with checkpoint on;
commit;
'''%(stage_schema, right_info, ftp_stage_absolute_filename, delimiter_type_id)
try:
iserr = 0
print *************begin to execute load table command...*************\n
if self.debug == 1:
self._printinfo(loadsql.strip())
#self.cursor.execute(loadsql.strip())
loadsql='''dbisql -c uid=%s;pwd=%s -host %s -port %s -nogui %s'''%(self.uid,self.pwd,sybaseiq_ipport[0],sybaseiq_ipport[1],loadsql)
os.system(loadsql)
print \n*************end to execute load table command...*************
print **************************successful**************************
except exception,err:
iserr = 1
print return value %s,error %s % (iserr,err)
return iserr
#main
def main():
# 检查传入参数个数
if len(sys.argv) print 'usage: python sybaseiq_loaddata.py sybasedsn username password ftp_id cur_static_time\n'
sys.exit(1)
# 定义连接sybase iq的信息
dbinfo = []
#dbinfo.append('sybaseiqdsn')
#dbinfo.append('jiang')
#dbinfo.append('jiang')
dbinfo.append(sys.argv[1])
dbinfo.append(sys.argv[2])
dbinfo.append(sys.argv[3])
ftp_id = sys.argv[4]
cur_static_time = sys.argv[5]
siq = sybaseiqload(dbinfo)
ret = siq.loaddata(ftp_id,cur_static_time)
return ret
if __name__ == '__main__':
sys.exit(main())