一、需求：
利用mongo元数据库中提供的资源描述，去校验csv中的每条数据
二、实现
首先配置好原数据，以及路径传递，还有jython模块
注意，jython有很多第三方包是没办法直接使用的，需要用sys去加载，这时我们会碰上一个最大的难题，就是第三方包的处理。由于jython是运行在jvm上的，所以，需要c语言运行环境的包在此时都无法调用成功，例如pandas，numpy等，但其他第三方包在sys成功加载后还是可以调用成功的，例如pymongo,要把这些包和其依赖包放在指定路径下。
streamsets最恶心的一点就是云端调试，问题与bug都要放在records.output中去打印输出
还有就是要非常注意streamsets本身的知识和结构,比如records是个list，而record是个对象；batch by batch 和record by record是两种不同的运行模式等，如何利用他们的性质进行编程仍是我们需要学习的
注意python格式的问题，循环的问题，还有业务逻辑处理的问题
三、编程
import sys
#sys.path.append('D:\JavaWorkplace\jython\jpython')
sys.path.append("/home/fengwenke/usr/streamset/jar/JPS.jar")
sys.path.append("/home/fengwenke/usr/streamset/python")
sys.setrecursionlimit(1000000)
from pymongo import MongoClient
import datetime as dt
import re
import json
conn = MongoClient('114.115.156.237', 27027)
db = conn.bigdata
db.authenticate("gwssi", "gwssi123")
res = db.resourceProfile
for record in records:
name = record.value['filepath']
#这个6和0是写死的，需要改，6需要根据csv路径的不同进行修改 0可能不需要改
csvName = name.split('/')[6]
tableName = csvName.split('_')[0]
a =list(res.find({"essentialInfo.resCode":tableName}))
meteData = []
for i in a:
for s in i['dataInfos']:
meteData.append(s['isPrimaryKey'])
meteData.append(s['dataName'])
meteData.append(s['dataType'])
print dt
newDate = dt.datetime.utcnow().strftime("%Y-%m-%d")
meteNameCollection = []
meteTypeCollection = []
meteIsprikeyCollection = []
#从mongo里拿出元数据的名字
for meteNameIndex in range(len(meteData)):
if (meteNameIndex+2)%3 ==0:
meteName = meteData[meteNameIndex]
meteNameCollection.append(meteName)
meteType = meteData[meteNameIndex+1]
meteTypeCollection.append(meteType)
meteIspri = meteData[meteNameIndex -1]
meteIsprikeyCollection.append(meteIspri)
dataNameCollection = []
for recordIndex in range(len(records)):
try:
# Create a string field to store the current date with the specified format
#record.value["3"] = meteData[8]
#从数据流里取出第一列
if recordIndex == 0:
#从第一列里拿出每个名字
for dataNameIndex in range(len(records[0].value)):
dataNameCollection.append(records[0].value['{0}'.format(dataNameIndex)])
else:
#利用这个数据匹配元数据，并对其他的数据类型进行校验 为什么不拆成两层循环，因为record记录会覆盖
for dataNameIndex2 in range(len(dataNameCollection)):
for meteNameIndex in range(len(meteNameCollection)):
if dataNameCollection[dataNameIndex2] == meteNameCollection[meteNameIndex]:
if meteIsprikeyCollection[meteNameIndex] == 1:
#读取对应的元数据类型. 时间。测试完成
if meteTypeCollection[meteNameIndex] == "timestamp":
matchRule = r'\d{4}(\-|\/|.)\d{1,2}\1\d{1,2}'
matchData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
if re.match(matchRule,matchData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
#读取对应的元数据类型. 字母数字混合数据。测试完成
if meteTypeCollection[meteNameIndex] == "varchar":
#字母数字混合数据
mixedData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
mixRule = '^(?=.*\d)(?=.*[a-zA-Z])(?=.*[\u4E00-\u9FA5])[\u4E00-\u9FA5A-Za-z0-9]*$'
rg = re.compile(mixRule,re.IGNORECASE|re.DOTALL)
mixJudge = rg.search(mixedData)
if mixJudge :
records[recordIndex].value['{0}'.format(dataNameIndex2)]= "true"
#英文
elif re.match('^[A-Za-z]+$',mixedData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
#中文
elif re.match(u"[\u4e00-\u9fa5]+",mixedData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
#空值
elif records[recordIndex].value['{0}'.format(dataNameIndex2)] == "":
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "zhujianweikong"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
#整数
if meteTypeCollection[meteNameIndex] == "integer" or meteTypeCollection[meteNameIndex] == "bigint":
matchRule = '^-?\\d+$'
matchData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
if re.match(matchRule,matchData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
#浮点数
if meteTypeCollection[meteNameIndex] == "float" or meteTypeCollection[meteNameIndex] == "double":
matchRule = '^(-?\\d+)(\\.\\d+)?$'
matchData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
if re.match(matchRule,matchData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
else:
#读取对应的元数据类型. 时间。测试完成
if meteTypeCollection[meteNameIndex] == "timestamp":
matchRule = r'\d{4}(\-|\/|.)\d{1,2}\1\d{1,2}'
matchData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
if re.match(matchRule,matchData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
#读取对应的元数据类型. 字母数字混合数据。测试完成
if meteTypeCollection[meteNameIndex] == "varchar":
#字母数字混合数据
mixedData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
mixRule = '^(?=.*\d)(?=.*[a-zA-Z])(?=.*[\u4E00-\u9FA5])[\u4E00-\u9FA5A-Za-z0-9]*$'
rg = re.compile(mixRule,re.IGNORECASE|re.DOTALL)
mixJudge = rg.search(mixedData)
if mixJudge :
records[recordIndex].value['{0}'.format(dataNameIndex2)]= "true"
#英文
elif re.match('^[A-Za-z]+$',mixedData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
#中文
elif re.match(u"[\u4e00-\u9fa5]+",mixedData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
#空值
elif records[recordIndex].value['{0}'.format(dataNameIndex2)] == "":
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
#整数
if meteTypeCollection[meteNameIndex] == "integer" or meteTypeCollection[meteNameIndex] == "bigint":
matchRule = '^-?\\d+$'
matchData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
if re.match(matchRule,matchData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
#浮点数
if meteTypeCollection[meteNameIndex] == "float" or meteTypeCollection[meteNameIndex] == "double":
matchRule = '^(-?\\d+)(\\.\\d+)?$'
matchData = records[recordIndex].value['{0}'.format(dataNameIndex2)]
if re.match(matchRule,matchData):
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "true"
else:
records[recordIndex].value['{0}'.format(dataNameIndex2)] = "false"
# Write record to processor output
output.write(records[recordIndex])
conn.close()
except Exception as e:
# Send record to error
error.write(records[recordIndex], str(e))
网友评论