5.10 Go语言文本大数据处理（1）：读入、清洗、分类

本篇文章IT兄弟连GO语言学院小美 给读者们分享一下5.10 Go语言文本大数据处理（1）：读入、清洗、分类 对GO语言感兴趣想要学习Golang开发技术的小伙伴就随小编来了解一下吧。 **需求分析**  - 如图是一个大小达到1.3G的纯文本大数据，其中记录了情人节当天帝各大宾馆的开房数据~ - 我们对这一大数据文本进行读取、清洗、分类、入库； **读取文本大数据** - 通过`ioutil.ReadFile(filename)`API可以傻瓜式地一次性读入全部数据 ``` //一次性将全部数据载入内存（不可取） func main011() { //一次性将全部数据载入内存（不可取） contentBytes, err := ioutil.ReadFile(`D:\GoIP\腾讯课堂公开课2019\数据\kaifangX.txt`) if err != nil { fmt.Println("读入失败", err) } contentStr := string(contentBytes) //以换行符为定界符，将大文本炸碎为行的切片 lineStrs := strings.Split(contentStr, "\n\r") //逐行打印 for _, lineStr := range lineStrs { //fmt.Println(lineStr) newStr, _ := ConvertEncoding(lineStr, "GBK", "UTF-8") fmt.Println(newStr) } } ``` - 然而考虑到数据总量在1G以上，一次性读入内存，不但增加内存压力，产生爆内存的风险，还会长时间阻塞，效率极低； - 正确的做法是，使用缓冲区进行逐行读取： ``` //缓冲式读取 func main012() { //打开文件 file, _ := os.Open(`D:\GoIP\腾讯课堂公开课2019\数据\kaifangX.txt`) defer file.Close() //创建文件的缓冲读取器 reader := bufio.NewReader(file) for { //逐行读取 lineBytes, _, err := reader.ReadLine() //读到文件末尾时退出循环 if err == io.EOF { break } //输出当前行内容 gbkStr := string(lineBytes) utfStr, _ := ConvertEncoding(gbkStr, "GBK", "UTF-8") fmt.Println(utfStr) } } ``` - 该大数据文本的原始内容是GBK编码，为了在Goland中便于输出查看，我们在输出时对其进行了编码转换，这里我们用到了一个第三方包`github.com/axgle/mahonia` ``` /* 将文本转换字符编码并返回 srcStr 待转码的原始字符串 srcEncoding 原始字符串的编码（字符集） dstEncoding 目标编码（字符集） */ func ConvertEncoding(srcStr string, srcEncoding string, dstEncoding string) (dstStr string, err error) { //创建指定字符集的解码器 srcDecoder := mahonia.NewDecoder(srcEncoding) dstDecoder := mahonia.NewDecoder(dstEncoding) //将内容转换为UTF-8字符串 utfStr := srcDecoder.ConvertString(srcStr) //将UTF-8字节转换为目标字符集的字节 _, dstBytes, err := dstDecoder.Translate([]byte(utfStr), true) if err != nil { return } //还原为字符串并返回 dstStr = string(dstBytes) return } ``` **数据清洗** - 数据的来源各式各样，数据质量良莠不齐，在入库之前，我们需要对数据进行必要的清洗工作； - 在本例中，部分数据的身份证号码信息是错误非法的（例如长度），我们通过清洗对这部分数据进行筛除； ``` func main021() { //打开原始文本 srcFile, _ := os.Open(`D:\GoIP\腾讯课堂公开课2019\数据\kaifangX.txt`) defer srcFile.Close() //准备一个优质数据文件 goodFile, _ := os.OpenFile(`D:\GoIP\腾讯课堂公开课2019\数据\kaifang_good.txt`, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) defer goodFile.Close() //准备一个不良数据文件 badFile, _ := os.OpenFile(`D:\GoIP\腾讯课堂公开课2019\数据\kaifang_bad.txt`, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644) defer badFile.Close() //创建原始文件的缓冲式读取器 reader := bufio.NewReader(srcFile) //循环读取 for { //一次读取一行数据 lineBytes, _, err := reader.ReadLine() //读到文件末尾时退出循环 if err == io.EOF { break } //转码得到当前行的UTF-8字符串 gbkStr := string(lineBytes) lineStr, _ := ConvertEncoding(gbkStr, "GBK", "UTF-8") //fmt.Println(lineStr) //使用逗号将当前行打散为子串 fields := strings.Split(lineStr, ",") //如果第二个子串的长度为18位，我们认为它是一个合法的身份证信息 if len(fields) > 1 && len(fields[1])==18{ //摘取到另一个优质的数据文件中 goodFile.WriteString(lineStr+"\n") fmt.Println("Good:",lineStr) }else{ //不合法的写入不良数据文件中 badFile.WriteString(lineStr+"\n") fmt.Println("Bad:",lineStr) } } } ``` - 技术栈主要是文件读写和字符串处理 **数据分类** - 此处我们根据身份证中的省份信息，将不同省份的开房者存入不同的文件中 - 主要技术栈依然包含文件读写和字符串处理 - 另外还用到了并发和管道，因为并发向34个省份文件中写入数据的效率显然是要高于一个线程的； 整体思路如下： - 主协程负责逐行读取文本大数据 - 另外开设34条子协程，负责对不同省份文件进行写入，从34个不同的管道中扫描数据并写出文件； - 主协程根据身份证号反映的不同省份，将读入的信息丢入不同的管道，由对应的子协程进行文件写出； - 当文件读取完毕时，关闭所有的数据管道（通知子协程停止数据扫描）； - 主协程通过等待组等待所有子协程完成任务； 代码如下： ``` package main import ( "bufio" "fmt" "io" "os" "strings" "sync" ) /*省份结构体*/ type Province struct { //省的Id（身份证号的前两位） Id string //省份名称 Name string //省份对应的文件 File *os.File //写入该文件的数据管道 chanData chan string } /*全局等待组*/ var ( wg sync.WaitGroup ) func main031() { //创建省份数据map，通过Id查询省信息 pMap := make(map[string]*Province) /* 为每个省创建一个文件 为每个省创建一个Province实例，丢入map */ ps := []string{"北京市11", "天津市12", "河北省13", "山西省14", "内蒙古自治区15", "辽宁省21", "吉林省22", "黑龙江省23", "上海市31", "江苏省32", "浙江省33", "安徽省34", "福建省35", "江西省36", "山东省37", "河南省41", "湖北省42", "湖南省43", "广东省44", "广西壮族自治区45", "海南省46", "重庆市50", "四川省51", "贵州省52", "云南省53", "西藏自治区54", "陕西省61", "甘肃省62", "青海省63", "宁夏回族自治区64", "新疆维吾尔自治区65", "台湾省71", "香港特别行政区81", "澳门特别行政区91",} for _, p := range ps { //截取省份名称和省份Id name := p[:len(p)-2] id := p[len(p)-2:] //创建省份对象，并丢入map province := Province{Id: id, Name: name} pMap[id] = &province //为每个省份关联一个预备写入的数据文件 file, _ := os.OpenFile(`D:\GoIP\腾讯课堂公开课2019\数据\`+province.Name+".txt", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) province.File = file defer file.Close() //为当前省份关联换一个数据管道 province.chanData = make(chan string, 100) } /*为每个省创建一条数据写入协程，从各自的管道中读取数据并写入对应的文件*/ for _, province := range pMap { //为当前省份创建一条数据写入协程 wg.Add(1) go func(p *Province) { //执行数据写入 writeFile(p) //标记协程结束 wg.Done() }(province) } /*主协程读入数据，将不同省份的记录丢入对应的管道*/ file, _ := os.Open(`D:\GoIP\腾讯课堂公开课2019\数据\kaifang_good.txt`) defer file.Close() //创建缓冲读取器 reader := bufio.NewReader(file) for { //读取一行 lineBytes, _, err := reader.ReadLine() //读取完毕时，关闭所有数据管道，并退出读取 if err == io.EOF { fmt.Println("已经读到文件末尾！") /*遍历关闭所有数据管道（以通知子协程停止数据扫描）*/ for _, province := range pMap { close(province.chanData) fmt.Println(province.Name, "管道已关闭", province.chanData) } break } //拿出省份ID lineStr := string(lineBytes) fieldsSlice := strings.Split(lineStr, ",") id := fieldsSlice[1][0:2] //根据Id查询得到省份，进而向其管道中写入当前行 if province, ok := pMap[id]; ok { province.chanData <- (lineStr + "\n") } else { //这里其实也是不合法的数据 fmt.Println("莫名其妙的省", id) } } //阻塞等待所有协程结束任务 wg.Wait() fmt.Println("main over!") } /*扫描指定省份的管道，取出数据并写出到对应的文件中*/ func writeFile(province *Province) { //扫描管道中的数据，管道关闭时循环结束 for lineStr := range province.chanData { province.File.WriteString(lineStr) //fmt.Print(province.Name, "写入", lineStr) } fmt.Println(province.Name, "管道遍历已结束", province.chanData) } ``` 想要了解更多关于GO语言开发方面内容的小伙伴， 请关注IT兄弟连官网、公众号：GO语言研习社， IT兄弟连教育有专业的微软、谷歌讲师为您指导， 此外IT兄弟连老师精心推出的GO语言教程定能让你快速掌握GO语言从入门到精通开发实战技能。