• 任務描述

????????對數據按照一定規則進行清洗。

????????清洗規則：

• 處理數據中的時間戳（秒級）將其轉化為" 年-月-日 時:分:秒 "這種格式；

• 處理數據中的省份編碼，結合mysql的表數據對應，將其轉換成省份名稱；

• 處理用戶手機號，與mysql的表數據對應，關聯用戶的真實姓名；

• 處理數據中的開始時間與結束時間并計算通信時長(以秒為單位)；

• 設置數據來源文件路徑及清洗后的數據存儲路徑： 數據來源路徑為： /user/test/input/a.txt （HDFS）； 清洗后的數據存放于：/user/test/output (HDFS)。

輸出格式：

鄧二,張倩,13666666666,15151889601,2018-03-29?10:58:12,2018-03-29?10:58:42,30,黑龍江省,上海市

• 代碼實現

1.dbhelper.py文件

# dbhelper.pyimport pymysql import sys import codecsclass DBHelper:def get_connection():# 根據題目提供的憑據建立到mysql服務器的連接"conn"，注意字符集指定為"utf8mb4"######## Begin ############conn = pymysql.connect(host='localhost',port=3306,\user='root',passwd='123123',\charset='utf8mb4',db='mydb')######## End ############ return conn@classmethoddef get_region(cls):conn = cls.get_connection()regions = dict()with conn.cursor() as cur:#從數據庫中查詢所有的省市代碼和省市名稱，并保存到字典regions中。############ Begin ###################cur.execute("select CodeNum,Address from allregion")for s in cur.fetchall():regions[s[0]] = s[1] ############ End #################conn.close()return regions@classmethoddef get_userphones(cls):conn = cls.get_connection()userphones = dict()with conn.cursor() as cur:#從數據庫中查詢所有的電話號碼和對應的姓名，并保存到字典userphones中。############ Begin ###################cur.execute("select phone,trueName from userphone")for t in cur.fetchall():userphones[t[0]] = t[1] ############ End #################conn.close()return userphonesdef main():sys.stdout = codecs.getwriter('utf-8')(sys.stdout.detach())region = DBHelper.get_region()users = DBHelper.get_userphones() if __name__ == '__main__':main()

2.mapper文件

#! /usr/bin/python3 # # mapper.py import sys from dbhelper import DBHelper import codecs import time# 獲取“省市代碼：省市名稱”項并保存在字典regions中； # 獲取“電話號碼：姓名”項并保存在字典userphones中。 regions = DBHelper.get_region() userphones = DBHelper.get_userphones()def main():# 正確輸出utf-8編碼的漢字sys.stdout = codecs.getwriter('utf-8')(sys.stdout.detach())for line in sys.stdin:line = line.strip()mapper(line)def mapper(line):# 輸出形如“鄧二,張倩,13666666666,15151889601,2018-03-29 10:58:12,2018-03-29 10:58:42,30,黑龍江省,上海市”的字符串# 本題不需要reduce階段，輸出題目要求的內容即可，不需要使用“鍵\t值”的形式。########## begin ##############items = line.split(',')caller = userphones.get(items[0])reciever = userphones.get(items[1])begin_time = int(items[2])end_time = int(items[3])caller_address = regions.get(items[4])reciever_address = regions.get(items[5])print(caller,reciever,sep=',',end=',')print(','.join(items[:2]),end=',')print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(begin_time)),end=',')print(time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(end_time)),end=',')print(str(end_time - begin_time),end=',')print(caller_address,reciever_address,sep=',')########### End #################if __name__ == '__main__':main()

總結

以上是生活随笔為你收集整理的3.2 电信数据清洗的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。