大数据分析的Python实战指南：数据处理、可视化与机器学习【上进小菜猪大数据】

import pandas as pd

# 导入原始数据
data = pd.read_csv('data.csv')

# 处理缺失值
data = data.dropna()  # 删除包含缺失值的行
data = data.fillna(0)  # 使用0填充缺失值

# 处理重复值
data = data.drop_duplicates()  # 删除重复的行

# 格式转换
data['date'] = pd.to_datetime(data['date'])  # 将日期列转换为日期格式

# 其他数据清洗操作，如数据类型转换、异常值处理等

import matplotlib.pyplot as plt
import seaborn as sns

# 统计特征
data.describe()

# 直方图
plt.hist(data['age'], bins=30)
plt.xlabel('Age')
plt.ylabel('Frequency')
plt.title('Age Distribution')
plt.show()

# 散点图
plt.scatter(data['income'], data['spending'])
plt.xlabel('Income')
plt.ylabel('Spending')
plt.title('Income vs. Spending')
plt.show()

# 热力图
correlation_matrix = data.corr()
sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm')
plt.title('Correlation Matrix')
plt.show()

# 其他数据探索和可视化操作，如箱线图、折线图等

from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score

# 准备特征和标签
X = data[['age', 'income']]
y = data['label']

# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 创建并训练模型
model = LogisticRegression()
model.fit(X_train, y_train)

# 在测试集上进行预测
y_pred = model.predict(X_test)

# 评估模型性能
accuracy = accuracy_score(y_test, y_pred)
print("Model Accuracy:", accuracy)

import dask.dataframe as dd

# 使用Dask加载大型数据集
data = dd.read_csv('big_data.csv')

# 数据转换和操作
data['new_column'] = data['column1'] + data['column2']
data = data[data['value'] > 0]

# 分组和聚合操作
grouped_data = data.groupby('category')['value'].sum()

# 并行计算
result = grouped_data.compute()

# 大数据处理和分布式计算的其他操作，如分区、合并、并行化等

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# 创建Spark Streaming上下文
sc = SparkContext("local[2]", "StreamingExample")
ssc = StreamingContext(sc, 1)

# 监听数据流
stream = ssc.socketTextStream("localhost", 9999)

# 数据处理和分析
processed_data = stream.filter(lambda x: len(x) > 0).flatMap(lambda x: x.split()).map(lambda x: (x, 1)).reduceByKey(lambda a, b: a + b)

# 打印结果
processed_data.pprint()

# 启动流式处理
ssc.start()
ssc.awaitTermination()

# 实时数据处理和流式分析的其他操作，如窗口操作、状态管理等

# 使用Apache Spark进行大数据处理
from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder.appName("DataProcessing").getOrCreate()

# 读取数据
data = spark.read.csv('big_data.csv', header=True, inferSchema=True)

# 数据处理和转换
processed_data = data.filter(data['value'] > 0).groupBy('category').sum('value')

# 显示结果
processed_data.show()

# 关闭SparkSession
spark.stop()

# 其他数据存储和大数据平台的使用示例，如HBase的数据存取、Kafka的数据流处理等