• 任務(wù)描述

????????對(duì)數(shù)據(jù)按照一定規(guī)則進(jìn)行清洗。

????????清洗規(guī)則：

• 處理數(shù)據(jù)中的時(shí)間戳（秒級(jí)）將其轉(zhuǎn)化為" 年-月-日 時(shí):分:秒 "這種格式；

• 處理數(shù)據(jù)中的省份編碼，結(jié)合mysql的表數(shù)據(jù)對(duì)應(yīng)，將其轉(zhuǎn)換成省份名稱(chēng)；

• 處理用戶(hù)手機(jī)號(hào)，與mysql的表數(shù)據(jù)對(duì)應(yīng)，關(guān)聯(lián)用戶(hù)的真實(shí)姓名；

• 處理數(shù)據(jù)中的開(kāi)始時(shí)間與結(jié)束時(shí)間并計(jì)算通信時(shí)長(zhǎng)(以秒為單位)；

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

輸出格式：

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

• 代碼實(shí)現(xiàn)

1.dbhelper.py文件

# dbhelper.pyimport pymysql import sys import codecsclass DBHelper:def get_connection():# 根據(jù)題目提供的憑據(jù)建立到mysql服務(wù)器的連接"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:#從數(shù)據(jù)庫(kù)中查詢(xún)所有的省市代碼和省市名稱(chēng)，并保存到字典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:#從數(shù)據(jù)庫(kù)中查詢(xún)所有的電話號(hào)碼和對(duì)應(yīng)的姓名，并保存到字典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# 獲取“省市代碼：省市名稱(chēng)”項(xiàng)并保存在字典regions中； # 獲取“電話號(hào)碼：姓名”項(xiàng)并保存在字典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階段，輸出題目要求的內(nèi)容即可，不需要使用“鍵\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()

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。