Python学习笔记（八）

首先需要明确：

Python的数据清洗不是大数据的解决方案！！

大数据通常使用MR或Spark进行数据的清洗！！

大数据的数据来源中以业务数据和网站日志为主!!!

（sqoop/Flume/NiFi/Kafka）

收据采集->>数据录入->>数据清洗->>数据处理->>数据集成->>数据监管->>数据分析->>数据服务

1. 了解数据采集的数据源与基本方法
2. 了解数据清洗的基本流程与方法
3. 掌握怎样用Python实施数据清洗
4. 掌握怎样用Python实施数据校验
5. 了解元数据并理解其在大数据环境中的重要作用
6. 了解数据存储、处理、集成、分析、服务等基本概论

数据采集

数据采集 确定数据需求 确定需要采集的数据字段 制定采集方法 验证采集数据的有效性 数据采集的关注点 数据有效性

数据来源

数据质量

• 数据质量是数据采集阶段最重要的

• 常见数据质量问题

  • 重复数据
  • 缺失数据
  • 数据关联性缺失
  • 非法数据
  • 字段填写错误
  • 数据格式不正确

• 数据质量判断原则 准确、完整、完备、有效、一致、格式统一、不重复

数据校验

• 数据校验验证数据集中的数据是否有效

  • 数据类型校验
  • 数据格式校验

• 数据校验的前提

  • 了解业务需求
  • 了解数据组成、结构及相关性

• 校验数据的方法

  • schema/meta-data/规则
  • 数据校验工具-SAS
  • 编写校验程序

• 数据校验会多次实施

数据校验工具-voluptuous

python>> pip install voluptuous
用Schema校验数据有效性

使用fillna的多种方式填充NaN值
使用interpolate()插值器填充NaN值,根据日期或时间按值等差填充
使用dropna删除包含缺失值的记录

• 异常值 合法但远离大部分数据的值

• 判断异常值 通过标准差计算确定异常值范围 标准差取值范围T，绝对值大于T的值 通过数据频率分布计算异常值范围 超出90%数值分布的值

• 修正异常值的影响 Winsorizing（温莎法）

  • T (such as 1.95) * StandardDeviation + Mean
  • 异常值修正为边界值+1或-1，体现不在边界内

9、Python集成spark

• 在linux上安装Anaconda，并配置环境变量
• 在linux上安装spark，必能配置环境变量：SPARK_HOME 和 SPARK_CONF_DIR
• 执行以下步骤

ipython
from notebook.auth import passwd
passwd()
#键入密码
#获取sha1值，复制
#rw
#sha1:0cc7d44db1b9:1ce93f146c1e0faaebf73740ca9db8ba90c7adde

cd~
jupyter notebook --generate-config
vi ./.jupyter/jupyter_notebook_config.py
#添加输入以下内容
c.NotebookApp.allow_root = True
c.NotebookApp.ip = ‘*‘
c.NotebookApp.open_browser = False
c.NotebookApp.password = ‘sha1:粘贴上一步复制的值‘
c.NotebookApp.port = 7070

cd~
vi /etc/profile
#添加以下内容
export PYSPARK_PYTHON=$ANACONDA_HOME/bin/python3
export PYSPARK_DRIVER_PYTHON=$ANACONDA_HOME/bin/jupyter
export PYSPARK_DRIVER_PYTHON_OPTS="notebook"
ipython_opts="notebook -pylab inline"

source /etc/profile

cd~
vi .jupyter/jupyter_notebook_config.py
#添加以下内容
c.NotebookApp.notebook_dir=‘自己定义的工作目录‘

使用notebook开发spark

• cmd: pyspark
• 浏览器连上jupyter 7070
• 进入notebook+spark的环境

在pyspark中可以直接使用spark，语法与scala类似，主要有以下不同之处

1、匿名函数写法：scala是直接写，py是lambda表达式

2、可迭代对象（列表、列）的取值符号，scala是()或[]，py可能是反的

使用pyspark解析复杂字段

from pyspark.sql.functions import *
from pyspark.sql.types import *
df = spark.read.option("header", "true").csv("file:///root/example/movies_metadata.csv")
# Define the schema for the movie category data field
genres = ArrayType(StructType([StructField("id", IntegerType(), False), StructField("name", StringType(), False)]))

# Organize the movie category with the original move id
df_MovieCategory = df.withColumn("movie_category", from_json(col("genres"), genres))   .select(col("id"), col("movie_category")).select(col("id"), explode(col("movie_category")))   .select(col("id"), col("col.name"))

Python学习笔记（八）

原文：https://www.cnblogs.com/whoyoung/p/11424212.html