python总结 + 部署简单项目 到生产
-> filter过滤：
list(filter(lambda x: x[0].find('tmp') == -1, table_temp_r))
-> 自定义map：
def map_format(x):
if(x.find('d_f_') == -1):
return 'd_f_artemis{}_{}'.format(data_f,x)
else:
return x
return list(map(lambda x: map_format(x), tables_p))
-> 字符串EL表达式：
"use {}".format(prod_base)
-> 列表、字典不为空的判断:
if l == []: if m == {}:
-> 异常处理：
try:
out_engine.execute("drop table if exists field_diff_database")
out_engine.execute("drop table if exists table_diff_database")
except Exception as e:
raise e
-> 获取时间：
time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())) #time.time() 是时间戳; time.localtime() 获取时间结构体; time.strftime() 格式化时间
-> IO处理:
pro_file = open(self.filename, 'Ur')
for line in pro_file.readlines():
line = line.strip().replace('\n', '')
字符串切分: 直接用[]
if line.find("#")!=-1:
line=line[0:line.find('#')]
获取文件地址:
path = os.getcwd()
file_path = os.path.join(path,'prod.cfg')
pandas:
指定表的顺序
colums = ['database','table','field']
df = pd.DataFrame(l,columns=colums)
df.to_sql('field_diff_database',out_engine,if_exists='append',index=False)
切片出对应的列
df.loc['database':'field']
df.loc[['database','field']]
使用index切片
df.iloc[1:3]
部署生产:
if __name__ == '__main__':
main_()
vim cron.txt
*/1 * * * * /data/anaconda/bin/python /home/hadoop/python_task/python_prod/prod.py > /home/hadoop/python_task/python_prod/result.log 2>&1
crontab -r / -l/ cron.txt
#crontab 不能使用os.getcwd() 不准确
path = os.path.dirname(os.path.abspath(__file__))
file_path = os.path.join(path,'data/prod.cfg')
