Apache Airflow大数据集成：Spark、Hadoop、Flink

【免费下载链接】airflow Apache Airflow - A platform to programmatically author, schedule, and monitor workflows 【免费下载链接】airflow 项目地址: https://gitcode.com/GitHub_Trending/airflo/airflow

概述

Apache Airflow作为业界领先的工作流编排平台，在大数据生态系统中扮演着至关重要的角色。通过与Spark、Hadoop、Flink等主流大数据技术的深度集成，Airflow能够构建端到端的数据处理流水线，实现复杂ETL作业的编排、调度和监控。

本文将深入探讨Airflow与三大主流大数据框架的集成方案，提供实战代码示例和最佳实践指南。

核心集成架构

mermaid

Spark集成详解

SparkSubmitOperator核心用法

SparkSubmitOperator是Airflow与Spark集成的主要接口，支持多种部署模式：

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

spark_etl_task = SparkSubmitOperator(
    task_id="spark_etl_job",
    application="/path/to/your-spark-app.jar",
    conn_id="spark_default",
    application_args=["--input", "hdfs://input/path", "--output", "hdfs://output/path"],
    jars="/path/to/dependency1.jar,/path/to/dependency2.jar",
    driver_memory="2g",
    executor_memory="4g",
    num_executors=4,
    executor_cores=2,
    verbose=True
)

参数配置表

参数 描述 示例值
application Spark应用JAR或Python文件路径 /app/spark-job.jar
conn_id Spark连接配置ID spark_standalone
application_args 应用命令行参数 ["--date", "2024-01-01"]
jars 依赖JAR包路径 hdfs:///jars/*.jar
files 需要分发的文件 config.properties
driver_memory Driver内存配置 2g
executor_memory Executor内存配置 4g
num_executors Executor数量 4
executor_cores 每个Executor核心数 2

Hadoop HDFS集成

HDFS传感器监控文件

from airflow.providers.apache.hdfs.sensors.hdfs import HdfsSensor

hdfs_file_sensor = HdfsSensor(
    task_id="wait_for_input_file",
    filepath="hdfs://namenode:8020/data/input/{{ ds }}.csv",
    hdfs_conn_id="hdfs_default",
    timeout=300,
    poke_interval=30
)

WebHDFS操作示例

from airflow.providers.apache.hdfs.hooks.webhdfs import WebHDFSHook

def process_hdfs_data(**kwargs):
    hook = WebHDFSHook(webhdfs_conn_id='hdfs_default')
    
    # 读取HDFS文件
    data = hook.read_file('/data/input/sample.csv')
    
    # 处理数据
    processed_data = process_data(data)
    
    # 写入HDFS
    hook.write_file('/data/output/processed.csv', processed_data, overwrite=True)
    
    return "Data processing completed"

Flink流处理集成

FlinkKubernetesOperator示例

from airflow.providers.apache.flink.operators.flink_kubernetes import FlinkKubernetesOperator

flink_streaming_job = FlinkKubernetesOperator(
    task_id="flink_stream_processing",
    application_file="/path/to/flink-job.yaml",
    flink_conn_id="flink_kubernetes",
    namespace="flink-namespace",
    deploy_mode="application",
    allow_non_restored_state=True
)

Flink作业YAML配置

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: stream-processing-job
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    parallelism.default: "2"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "4096m"
      cpu: 2
  job:
    jarURI: local:///opt/flink/usrlib/streaming-job.jar
    entryClass: com.example.StreamProcessingJob
    args: ["--kafka.topic", "input-topic", "--output.path", "hdfs://output/"]
    parallelism: 4
    upgradeMode: stateless

完整大数据流水线示例

from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.providers.apache.hdfs.sensors.hdfs import HdfsSensor
from airflow.providers.apache.flink.operators.flink_kubernetes import FlinkKubernetesOperator
from airflow.operators.python import PythonOperator
from airflow.operators.dummy import DummyOperator

default_args = {
    'owner': 'data_engineering',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3
}

with DAG(
    'big_data_processing_pipeline',
    default_args=default_args,
    description='End-to-end big data processing pipeline',
    schedule_interval='@daily',
    start_date=days_ago(1),
    tags=['spark', 'hadoop', 'flink']
) as dag:

    start = DummyOperator(task_id='start')
    
    # 监控HDFS输入文件
    wait_for_data = HdfsSensor(
        task_id='wait_for_input_data',
        filepath='hdfs:///data/raw/{{ ds }}/',
        timeout=3600,
        poke_interval=300
    )
    
    # Spark批处理
    spark_processing = SparkSubmitOperator(
        task_id='spark_batch_processing',
        application='hdfs:///apps/spark-etl.jar',
        application_args=[
            '--input', 'hdfs:///data/raw/{{ ds }}/',
            '--output', 'hdfs:///data/processed/{{ ds }}/'
        ],
        conn_id='spark_yarn',
        executor_memory='8g',
        num_executors=10
    )
    
    # Flink流处理
    flink_streaming = FlinkKubernetesOperator(
        task_id='flink_real_time_processing',
        application_file='hdfs:///configs/flink-streaming.yaml',
        flink_conn_id='flink_kubernetes'
    )
    
    # 数据质量检查
    data_quality_check = PythonOperator(
        task_id='data_quality_validation',
        python_callable=validate_data_quality,
        op_kwargs={'date': '{{ ds }}'}
    )
    
    end = DummyOperator(task_id='end')
    
    # 定义依赖关系
    start >> wait_for_data >> spark_processing
    spark_processing >> flink_streaming
    flink_streaming >> data_quality_check >> end

监控与错误处理

自定义监控指标

from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults
from airflow.providers.apache.spark.hooks.spark_submit import SparkSubmitHook

class MonitoredSparkOperator(BaseOperator):
    @apply_defaults
    def __init__(self, application, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.application = application
        
    def execute(self, context):
        hook = SparkSubmitHook()
        
        try:
            # 提交Spark作业
            self.log.info(f"Submitting Spark application: {self.application}")
            driver_id = hook.submit(self.application)
            
            # 监控作业状态
            while True:
                status = hook.get_status(driver_id)
                if status in ['FINISHED', 'FAILED', 'KILLED']:
                    break
                self.log.info(f"Job status: {status}")
                time.sleep(30)
                
            if status == 'FINISHED':
                self.log.info("Spark job completed successfully")
                return True
            else:
                raise Exception(f"Spark job failed with status: {status}")
                
        except Exception as e:
            self.log.error(f"Spark job execution failed: {str(e)}")
            # 发送告警通知
            send_alert_notification(
                f"Spark job {self.application} failed",
                str(e)
            )
            raise

性能优化最佳实践

资源调优配置表

组件 配置项 推荐值 说明
Spark spark.executor.memory 8-16g 根据数据量调整
spark.executor.cores 4-8 避免资源竞争
spark.sql.shuffle.partitions 200-400 优化shuffle性能
HDFS dfs.block.size 128-256MB 大文件处理优化
io.file.buffer.size 131072 提高IO性能
Flink taskmanager.memory.process.size 4-8g 流处理内存配置
parallelism.default 4-16 并行度设置

数据分区策略

# 优化Spark数据分区
def optimize_spark_partitions():
    from pyspark.sql import SparkSession
    
    spark = SparkSession.builder \
        .appName("OptimizedETL") \
        .config("spark.sql.adaptive.enabled", "true") \
        .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
        .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB") \
        .getOrCreate()
    
    # 读取数据并优化分区
    df = spark.read.parquet("hdfs:///data/input/")
    optimized_df = df.repartition(200)  # 根据数据量调整分区数
    
    return optimized_df

安全与权限管理

Kerberos认证集成

from airflow.providers.apache.hdfs.hooks.hdfs import HDFSHook

def secure_hdfs_operation():
    # 配置Kerberos认证
    hook = HDFSHook(
        hdfs_conn_id='hdfs_kerberos',
        proxy_user='data_engineer',
        kerberos=True
    )
    
    # 执行安全操作
    with hook.get_conn() as hdfs:
        if hdfs.exists('/secure/data'):
            content = hdfs.read('/secure/data/file.txt')
            return content.decode('utf-8')

总结

Apache Airflow与Spark、Hadoop、Flink的集成为大数据处理提供了完整的解决方案。通过合理的架构设计、性能优化和安全配置，可以构建出高效、可靠的大数据流水线。

关键要点：

  • 灵活的任务编排：支持批处理和流处理的混合工作流
  • 完善的监控体系：实时跟踪作业状态和性能指标
  • 资源优化：根据业务需求动态调整计算资源
  • 安全可靠：集成企业级安全认证和权限管理

通过本文提供的示例和最佳实践，您可以快速构建属于自己的大数据处理平台，实现数据价值的最大化。

【免费下载链接】airflow Apache Airflow - A platform to programmatically author, schedule, and monitor workflows 【免费下载链接】airflow 项目地址: https://gitcode.com/GitHub_Trending/airflo/airflow

Logo
ModelEngine社区

更多推荐

从崩溃到自愈：Temporal+MCP-Agent构建永不中断的AI工作流

你是否遇到过AI工作流执行到90%突然崩溃的绝望？训练了数小时的模型因服务器重启前功尽弃？团队协作时因某人未及时审批导致整个流程停滞？本文将展示如何通过Temporal与MCP-Agent的深度集成，彻底解决这些痛点，构建真正企业级的可靠AI工作流系统。读完本文你将掌握：- 3行代码实现工作流故障自愈- 零停机升级正在运行的AI任务- 跨团队协作的工作流信号机制- 从本地开发到生产集群

avatar ModelEngine社区
cover

从"数据喂养"到"数据工程"：大模型时代的数据哲学

avatar ModelEngine社区
cover

GPU 虚拟化：从理论到落地（一）三维正交坐标系讲透分类

avatar ModelEngine社区