数据处理专题（十三）

学会基本的图像处理技术。‍

OpenCV 基础

实践：使用 OpenCV 进行图像读取、显示和基本处理‍

03

代码示例

1. 导入必要的库

import cv2import numpy as npimport matplotlib.pyplot as plt

2. 图像读取

# 读取图像image_path = 'path_to_your_image.jpg'  # 替换为你的图像路径image = cv2.imread(image_path)# 检查图像是否成功读取if image is None:    print("图像读取失败，请检查路径是否正确。")else:    print("图像读取成功！")

3. 图像显示

# 使用 OpenCV 显示图像cv2.imshow('原图', image)cv2.waitKey(0)cv2.destroyAllWindows()# 使用 Matplotlib 显示图像plt.imshow(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))plt.title('原图')plt.axis('off')plt.show()

4. 图像基本信息

# 获取图像的基本信息height, width, channels = image.shapeprint(f"图像高度: {height} 像素")print(f"图像宽度: {width} 像素")print(f"图像通道数: {channels}")

5. 图像灰度化​​​​​​​

# 将图像转换为灰度图像gray_image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)# 显示灰度图像cv2.imshow('灰度图', gray_image)cv2.waitKey(0)cv2.destroyAllWindows()# 使用 Matplotlib 显示灰度图像plt.imshow(gray_image, cmap='gray')plt.title('灰度图')plt.axis('off')plt.show()

6. 图像裁剪​​​​​​​

# 裁剪图像cropped_image = image[100:400, 100:400]# 显示裁剪后的图像cv2.imshow('裁剪图', cropped_image)cv2.waitKey(0)cv2.destroyAllWindows()# 使用 Matplotlib 显示裁剪后的图像plt.imshow(cv2.cvtColor(cropped_image, cv2.COLOR_BGR2RGB))plt.title('裁剪图')plt.axis('off')plt.show()

7. 图像缩放​​​​​​​

# 缩放图像resized_image = cv2.resize(image, (width // 2, height // 2))# 显示缩放后的图像cv2.imshow('缩放图', resized_image)cv2.waitKey(0)cv2.destroyAllWindows()# 使用 Matplotlib 显示缩放后的图像plt.imshow(cv2.cvtColor(resized_image, cv2.COLOR_BGR2RGB))plt.title('缩放图')plt.axis('off')plt.show()

8. 图像旋转​​​​​​​

# 旋转图像center = (width // 2, height // 2)angle = 45scale = 1.0rotation_matrix = cv2.getRotationMatrix2D(center, angle, scale)rotated_image = cv2.warpAffine(image, rotation_matrix, (width, height))# 显示旋转后的图像cv2.imshow('旋转图', rotated_image)cv2.waitKey(0)cv2.destroyAllWindows()# 使用 Matplotlib 显示旋转后的图像plt.imshow(cv2.cvtColor(rotated_image, cv2.COLOR_BGR2RGB))plt.title('旋转图')plt.axis('off')plt.show()

9. 图像翻转​​​​​​​

# 翻转图像flipped_image = cv2.flip(image, 1)  # 1 表示水平翻转，0 表示垂直翻转，-1 表示水平和垂直翻转# 显示翻转后的图像cv2.imshow('翻转图', flipped_image)cv2.waitKey(0)cv2.destroyAllWindows()# 使用 Matplotlib 显示翻转后的图像plt.imshow(cv2.cvtColor(flipped_image, cv2.COLOR_BGR2RGB))plt.title('翻转图')plt.axis('off')plt.show()

10. 图像保存​​​​​​​

# 保存处理后的图像output_path = 'processed_image.jpg'cv2.imwrite(output_path, flipped_image)print(f"处理后的图像已保存到 {output_path}")

04

实践​​​​​​​

import cv2import numpy as npimport matplotlib.pyplot as plt# 读取图像image_path = 'path_to_your_image.jpg'  # 替换为你的图像路径image = cv2.imread(image_path)# 检查图像是否成功读取if image is None:    print("图像读取失败，请检查路径是否正确。")else:    print("图像读取成功！")    # 使用 OpenCV 显示图像    cv2.imshow('原图', image)    cv2.waitKey(0)    cv2.destroyAllWindows()    # 使用 Matplotlib 显示图像    plt.imshow(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))    plt.title('原图')    plt.axis('off')    plt.show()    # 获取图像的基本信息    height, width, channels = image.shape    print(f"图像高度: {height} 像素")    print(f"图像宽度: {width} 像素")    print(f"图像通道数: {channels}")    # 将图像转换为灰度图像    gray_image = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)    # 显示灰度图像    cv2.imshow('灰度图', gray_image)    cv2.waitKey(0)    cv2.destroyAllWindows()    # 使用 Matplotlib 显示灰度图像    plt.imshow(gray_image, cmap='gray')    plt.title('灰度图')    plt.axis('off')    plt.show()    # 裁剪图像    cropped_image = image[100:400, 100:400]    # 显示裁剪后的图像    cv2.imshow('裁剪图', cropped_image)    cv2.waitKey(0)    cv2.destroyAllWindows()    # 使用 Matplotlib 显示裁剪后的图像    plt.imshow(cv2.cvtColor(cropped_image, cv2.COLOR_BGR2RGB))    plt.title('裁剪图')    plt.axis('off')    plt.show()    # 缩放图像    resized_image = cv2.resize(image, (width // 2, height // 2))    # 显示缩放后的图像    cv2.imshow('缩放图', resized_image)    cv2.waitKey(0)    cv2.destroyAllWindows()    # 使用 Matplotlib 显示缩放后的图像    plt.imshow(cv2.cvtColor(resized_image, cv2.COLOR_BGR2RGB))    plt.title('缩放图')    plt.axis('off')    plt.show()    # 旋转图像    center = (width // 2, height // 2)    angle = 45    scale = 1.0    rotation_matrix = cv2.getRotationMatrix2D(center, angle, scale)    rotated_image = cv2.warpAffine(image, rotation_matrix, (width, height))    # 显示旋转后的图像    cv2.imshow('旋转图', rotated_image)    cv2.waitKey(0)    cv2.destroyAllWindows()    # 使用 Matplotlib 显示旋转后的图像    plt.imshow(cv2.cvtColor(rotated_image, cv2.COLOR_BGR2RGB))    plt.title('旋转图')    plt.axis('off')    plt.show()    # 翻转图像    flipped_image = cv2.flip(image, 1)  # 1 表示水平翻转，0 表示垂直翻转，-1 表示水平和垂直翻转    # 显示翻转后的图像    cv2.imshow('翻转图', flipped_image)    cv2.waitKey(0)    cv2.destroyAllWindows()    # 使用 Matplotlib 显示翻转后的图像    plt.imshow(cv2.cvtColor(flipped_image, cv2.COLOR_BGR2RGB))    plt.title('翻转图')    plt.axis('off')    plt.show()    # 保存处理后的图像    output_path = 'processed_image.jpg'    cv2.imwrite(output_path, flipped_image)    print(f"处理后的图像已保存到 {output_path}")

05

总结

通过今天的练习，你应该已经学会了如何使用 OpenCV 进行基本的图像处理，包括图像读取、显示、灰度化、裁剪、缩放、旋转和翻转

数据管道

学会安装和配置 Apache Airflow

使用 Airflow 构建一个简单的数据处理管道

调度和监控任务‍

01

目标

学会安装和配置 Apache Airflow

使用 Airflow 构建一个简单的数据处理管道

调度和监控任务‍

02

步骤

1. 安装 Apache Airflow

首先，我们需要安装 Apache Airflow。可以使用 pip 来安装：

pip install apache-airflow

2. 初始化 Airflow

初始化 Airflow 的数据库和配置文件：

airflow db init

3. 启动 Airflow Web 服务器

启动 Airflow 的 Web 服务器，这样我们可以在浏览器中监控和管理任务：

airflow webserver --port 8080

4. 启动 Airflow Scheduler

启动 Airflow 的调度器，它负责调度任务：

airflow scheduler

5. 创建 DAG 文件

DAG（Directed Acyclic Graph）是 Airflow 中的核心概念，表示一系列任务的依赖关系。我们将在 dags 目录下创建一个 Python 文件来定义我们的 DAG。

假设我们有一个简单的数据处理管道，包括以下几个任务：

从 CSV 文件中读取数据

清洗数据（处理缺失值和重复行）

按部门分组并计算每组的销售额均值

将结果保存到新的 CSV 文件

创建一个名为 simple_data_pipeline.py 的文件，内容如下：​​​​​​​

from airflow import DAGfrom airflow.operators.python_operator import PythonOperatorfrom datetime import datetime, timedeltaimport pandas as pdimport os# 定义默认参数default_args = {    'owner': 'airflow',    'depends_on_past': False,    'start_date': datetime(2023, 1, 1),    'retries': 1,    'retry_delay': timedelta(minutes=5),}# 创建 DAGdag = DAG(    'simple_data_pipeline',    default_args=default_args,    description='一个简单的数据处理管道',    schedule_interval=timedelta(days=1),)# 定义任务函数def read_csv_file():    """从 CSV 文件中读取数据"""    file_path = '/path/to/sales_data.csv'    df = pd.read_csv(file_path, encoding='utf-8-sig')    print(f"原始数据集: \n{df.head()}")    return dfdef clean_data(**context):    """清洗数据（处理缺失值和重复行）"""    df = context['ti'].xcom_pull(task_ids='read_csv_file')    # 检查每列的缺失值数量    missing_values = df.isnull().sum()    print(f"每列的缺失值数量: \n{missing_values}")    # 删除含有缺失值的行    df_cleaned = df.dropna()    print(f"删除缺失值后的数据集: \n{df_cleaned.head()}")    # 检查重复行    duplicates = df_cleaned.duplicated()    print(f"重复行: \n{duplicates}")    # 删除重复行    df_no_duplicates = df_cleaned.drop_duplicates()    print(f"删除重复行后的数据集: \n{df_no_duplicates.head()}")    return df_no_duplicatesdef group_and_calculate_mean(**context):    """按部门分组并计算每组的销售额均值"""    df = context['ti'].xcom_pull(task_ids='clean_data')    # 按 '部门' 列分组    grouped_by_department = df.groupby('部门')    # 计算每组的销售额均值    mean_sales_by_department = grouped_by_department['总价'].mean()    print(f"按 '部门' 列分组后，每组的销售额均值: \n{mean_sales_by_department}")    return mean_sales_by_departmentdef save_results(**context):    """将结果保存到新的 CSV 文件"""    mean_sales_by_department = context['ti'].xcom_pull(task_ids='group_and_calculate_mean')    result_path = '/path/to/mean_sales_by_department.csv'    mean_sales_by_department.to_csv(result_path, encoding='utf-8-sig')    print(f"结果已保存到 {result_path}")# 定义任务read_csv_task = PythonOperator(    task_id='read_csv_file',    python_callable=read_csv_file,    dag=dag,)clean_data_task = PythonOperator(    task_id='clean_data',    python_callable=clean_data,    provide_context=True,    dag=dag,)group_and_calculate_mean_task = PythonOperator(    task_id='group_and_calculate_mean',    python_callable=group_and_calculate_mean,    provide_context=True,    dag=dag,)save_results_task = PythonOperator(    task_id='save_results',    python_callable=save_results,    provide_context=True,    dag=dag,)# 设置任务依赖关系read_csv_task >> clean_data_task >> group_and_calculate_mean_task >> save_results_task

6. 配置 Airflow

确保 dags 目录在 Airflow 的配置文件中正确设置。通常情况下，Airflow 会自动检测 dags 目录下的 DAG 文件。

7. 运行和监控任务

打开浏览器，访问 http://localhost:8080，登录 Airflow 的 Web 界面。

在 DAG 列表中找到 simple_data_pipeline，点击进入详细页面。

点击 "Trigger Dag" 按钮手动触发任务，或者等待调度器自动运行任务。

在任务详情页面中，可以查看每个任务的运行状态和日志。‍

03

总结

通过今天的实践，你应该已经学会了如何使用 Apache Airflow 构建一个简单的数据处理管道。这个管道包括从 CSV 文件中读取数据、清洗数据、按部门分组计算销售额均值，并将结果保存到新的 CSV 文件。