#apachespark

Google Cloud is well positioned to provide enterprises with a unified, intelligent, open, and secure data and AI cloud. Dataproc, Dataflow, BigQuery, BigLake, and Vertex AI are used by thousands of clients in many industries across the globe for data-to-AI operations. From data intake and preparation to analysis, exploration, and visualization to ML training and inference, it presents BigQuery Studio, a unified, collaborative workspace for Google Cloud’s data analytics suite that speeds up data to AI workflows. It enables data professionals to:

Utilize BigQuery’s built-in SQL, Python, Spark, or natural language capabilities to leverage code assets across Vertex AI and other products for specific workflows.

Improve cooperation by applying best practices for software development, like CI/CD, version history, and source control, to data assets.

Enforce security standards consistently and obtain governance insights within BigQuery by using data lineage, profiling, and quality.

The following features of BigQuery Studio assist you in finding, examining, and drawing conclusions from data in BigQuery:

Code completion, query validation, and byte processing estimation are all features of this powerful SQL editor.

Colab Enterprise-built embedded Python notebooks. Notebooks come with built-in support for BigQuery DataFrames and one-click Python development runtimes.

You can create stored Python procedures for Apache Spark using this PySpark editor.

Dataform-based asset management and version history for code assets, including notebooks and stored queries.

Gemini generative AI (Preview)-based assistive code creation in notebooks and the SQL editor.

Dataplex includes for data profiling, data quality checks, and data discovery.

The option to view work history by project or by user.

The capability of exporting stored query results for use in other programs and analyzing them by linking to other tools like Looker and Google Sheets.

Follow the guidelines under Enable BigQuery Studio for Asset Management to get started with BigQuery Studio. The following APIs are made possible by this process:

To use Python functions in your project, you must have access to the Compute Engine API.

Code assets, such as notebook files, must be stored via the Dataform API.

In order to run Colab Enterprise Python notebooks in BigQuery, the Vertex AI API is necessary.

Single interface for all data teams

Analytics experts must use various connectors for data intake, switch between coding languages, and transfer data assets between systems due to disparate technologies, which results in inconsistent experiences. The time-to-value of an organization’s data and AI initiatives is greatly impacted by this.

By providing an end-to-end analytics experience on a single, specially designed platform, BigQuery Studio tackles these issues. Data engineers, data analysts, and data scientists can complete end-to-end tasks like data ingestion, pipeline creation, and predictive analytics using the coding language of their choice with its integrated workspace, which consists of a notebook interface and SQL (powered by Colab Enterprise, which is in preview right now).

For instance, data scientists and other analytics users can now analyze and explore data at the petabyte scale using Python within BigQuery in the well-known Colab notebook environment. The notebook environment of BigQuery Studio facilitates data querying and transformation, autocompletion of datasets and columns, and browsing of datasets and schema. Additionally, Vertex AI offers access to the same Colab Enterprise notebook for machine learning operations including MLOps, deployment, and model training and customisation.

Additionally, BigQuery Studio offers a single pane of glass for working with structured, semi-structured, and unstructured data of all types across cloud environments like Google Cloud, AWS, and Azure by utilizing BigLake, which has built-in support for Apache Parquet, Delta Lake, and Apache Iceberg.

One of the top platforms for commerce, Shopify, has been investigating how BigQuery Studio may enhance its current BigQuery environment.

Maximize productivity and collaboration

By extending software development best practices like CI/CD, version history, and source control to analytics assets like SQL scripts, Python scripts, notebooks, and SQL pipelines, BigQuery Studio enhances cooperation among data practitioners. To ensure that their code is always up to date, users will also have the ability to safely link to their preferred external code repositories.

BigQuery Studio not only facilitates human collaborations but also offers an AI-powered collaborator for coding help and contextual discussion. BigQuery’s Duet AI can automatically recommend functions and code blocks for Python and SQL based on the context of each user and their data. The new chat interface eliminates the need for trial and error and document searching by allowing data practitioners to receive specialized real-time help on specific tasks using natural language.

Unified security and governance

By assisting users in comprehending data, recognizing quality concerns, and diagnosing difficulties, BigQuery Studio enables enterprises to extract reliable insights from reliable data. To assist guarantee that data is accurate, dependable, and of high quality, data practitioners can profile data, manage data lineage, and implement data-quality constraints. BigQuery Studio will reveal tailored metadata insights later this year, such as dataset summaries or suggestions for further investigation.

Additionally, by eliminating the need to copy, move, or exchange data outside of BigQuery for sophisticated workflows, BigQuery Studio enables administrators to consistently enforce security standards for data assets. Policies are enforced for fine-grained security with unified credential management across BigQuery and Vertex AI, eliminating the need to handle extra external connections or service accounts. For instance, Vertex AI’s core models for image, video, text, and language translations may now be used by data analysts for tasks like sentiment analysis and entity discovery over BigQuery data using straightforward SQL in BigQuery, eliminating the need to share data with outside services.

Hire Scala Developers from Litoust India for AI, machine learning, database development, game applications and social media platform optimisation. Our Scala developers combine Apache Spark, Kafka, Akka and Scala.js to deliver scalable and high-performance software solutions. Our focus areas include predictive analytics, cognitive data processing, SQL database modifications, API development and enterprise application enhancements that improve speed, scalability, reporting capabilities, and overall business efficiency.

Get more:

In the age of big data, speed, scalability, and performance are essential — and Apache Spark delivers all three.

Apache Spark is a powerful unified analytics engine designed for large-scale data processing, enabling organizations to build high-performance batch and real-time streaming pipelines using distributed computing.

From ETL workflows and machine learning to real-time analytics, data engineering, and cloud-scale applications, Spark helps businesses process massive datasets efficiently and at scale.

With its ability to handle both batch and streaming workloads in a single platform, Apache Spark has become a core technology for modern data ecosystems.

Big data demands powerful processing — and Spark powers the future of scalable analytics.

Read more:

As data volumes continue to grow, organizations need faster and more scalable ways to analyze massive datasets — and Spark SQL is leading the way.

Built on Apache Spark, Spark SQL enables high-performance distributed query processing for large-scale analytics, data engineering, and real-time business intelligence.

From structured data analysis and ETL pipelines to machine learning integration and cloud-scale processing, Spark SQL helps businesses extract insights efficiently across complex data environments.

By combining the power of SQL with distributed computing, Spark SQL makes big data analytics faster, smarter, and more scalable.

In the era of big data, scalable analytics is the key to smarter decisions.

Read more:

Introduction

In the realm of academic projects, especially those focused on IT and data science, students often encounter the formidable challenge of managing and analyzing vast amounts of data—what we commonly refer to as "big data." Two of the most prominent frameworks that have emerged to tackle these challenges are Apache Spark and Hadoop. These tools are frequently chosen by students for their ability to handle large-scale data processing and analytics efficiently. However, the decision of which framework to utilize can be daunting, with each offering distinct advantages tailored to specific needs. Understanding the key differences between Apache Spark and Hadoop is crucial for students embarking on big data projects.

Overview of Apache Spark

Apache Spark is a fast, in-memory data processing engine known for its speed and ease of use. It provides a comprehensive suite of libraries for tasks such as SQL, streaming, machine learning, and graph processing, making it a versatile tool for various academic projects.

Speed and In-Memory Processing

One of Spark's standout features is its in-memory processing capability, which allows it to store intermediate data in memory rather than writing it to disk. This significantly enhances its processing speed, making it ideal for iterative tasks and real-time data analysis. For students working on projects requiring rapid computations, Spark provides a notable advantage.

Academic Use Cases

In the academic setting, Apache Spark is often used for projects involving:

Real-time data analysis: Due to its speed, Spark is an excellent choice for projects that require real-time insights.

Machine learning experiments: Spark's MLlib library supports various machine learning algorithms, facilitating quick experimentation.

Interactive data exploration: Spark's ability to process data quickly allows students to interactively explore large datasets, making it a favorite for data analysis assignments.

Overview of Hadoop

Hadoop is a well-established framework that includes components such as the Hadoop Distributed File System (HDFS) and MapReduce, which are central to its operation.

Batch Processing and Storage

Hadoop is designed for batch processing and excels at storing and processing large datasets across distributed computing environments. Its robust storage capabilities make it a strong candidate for projects that require processing historical data in large volumes.

Academic Fit

For student projects, Hadoop is particularly useful in scenarios such as:

Large-scale batch processing: Projects that involve processing significant amounts of historical data benefit from Hadoop's distributed storage and processing capabilities.

Data warehousing: Hadoop's scalability and storage efficiency make it suitable for creating data warehouses for academic research.

Data integration tasks: The framework is well-suited for integrating data from multiple sources.

Key Differences Between Spark and Hadoop

When comparing Spark and Hadoop, several factors come into play, including performance, ease of learning, and use cases.

Performance and Speed

Apache Spark: Known for its high speed due to in-memory processing, making it suitable for real-time and iterative tasks.

Hadoop: While not as fast as Spark due to its reliance on disk storage, it is highly efficient for batch processing.

Ease of Learning

Apache Spark: Offers a user-friendly API and supports multiple languages, including Python, Java, and Scala, making it accessible for beginners.

Hadoop: Requires understanding of Java-based MapReduce, which can be more challenging for students new to programming.

Programming Complexity

Apache Spark: Simpler to implement complex data processing tasks due to its higher-level abstractions.

Hadoop: Involves more boilerplate code and is less intuitive for complex operations.

Use Cases in Academic Projects

Spark: Best for projects needing quick iterations, real-time processing, or machine learning integration.

Hadoop: Suitable for handling extensive batch processing tasks and data storage needs.

Resource Requirements

Apache Spark: Requires more memory, which can be a limitation for students with constrained resources.

Hadoop: More efficient in terms of storage but demands significant disk space for data.

Which One Should Students Choose?

Deciding between Spark and Hadoop depends on several factors:

Assignment Requirements: Projects demanding real-time processing or rapid iteration may benefit from Spark, while those focusing on large-scale data storage and analysis might favor Hadoop.

Data Size: For projects with massive datasets, Hadoop's storage capabilities are advantageous.

Project Deadlines: Tight deadlines can be better managed with Spark's quicker processing times.

Learning Curve: Students new to big data might find Spark easier to learn due to its simplified API.

Common Use Cases in Academic Projects

Both frameworks offer distinct advantages for various academic applications:

Data Analysis Assignments: Spark's speed is beneficial for fast data exploration, while Hadoop excels in processing comprehensive datasets.

Machine Learning Projects: Spark's MLlib library makes it a go-to for machine learning tasks.

Log File Analysis: Hadoop's distributed computing capabilities are well-suited for analyzing extensive log files.

Real-time vs Batch Processing Examples: Spark is preferred for real-time projects, whereas Hadoop is ideal for batch processing.

Challenges Students Face

Students may encounter several challenges when working with these frameworks:

Installation and Setup Issues: Both tools require careful setup, which can be daunting for beginners.

Debugging Errors: Complex data processing tasks can lead to challenging debugging scenarios.

Limited System Resources: Spark's memory requirements may strain systems with limited resources.

Time Constraints: Learning and implementing these frameworks within project deadlines can be stressful.

How Expert Academic Support Can Help

Professional academic guidance can significantly ease the burden of working with big data tools. Services like PenContentDigital offer expert advice and support, ensuring plagiarism-free, on-time delivery of assignments. This support allowTo enhance the understanding of how Apache Spark and Hadoop work in academic projects, it's beneficial to provide some example code snippets to illustrate their usage. These examples will give students a practical perspective on how to implement these frameworks in their projects.

Apache Spark Code Example

Here’s a simple example of using Apache Spark to process a dataset. Assume we have a dataset of students' grades, and we want to calculate the average grade.

from pyspark.sql import SparkSession

Create a Spark session

spark = SparkSession.builder \ .appName("Grade Average Calculator") \ .getOrCreate()

Load the dataset

data = [("Alice", 85), ("Bob", 78), ("Cathy", 92), ("David", 88)] columns = ["Name", "Grade"]

Create a DataFrame

df = spark.createDataFrame(data, columns)

Calculate the average grade

average_grade = df.groupBy().avg("Grade").collect()[0][0]

print(f"The average grade is: {average_grade}")

Stop the Spark session

spark.stop()

This example demonstrates how to set up a Spark session, load data into a DataFrame, perform a group-by operation to calculate the average, and output the result.

Hadoop MapReduce Code Example

For Hadoop, here is a simple MapReduce example in Java to count the number of occurrences of each grade in a dataset.

import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class GradeCount {

public static class TokenizerMapper extends Mapper { private final static IntWritable one = new IntWritable(1); private Text grade = new Text();public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split(","); grade.set(fields[1]); // Assuming grade is the second field context.write(grade, one); }

}

public static class IntSumReducer extends Reducer { private IntWritable result = new IntWritable();public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); }

}

public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "grade count"); job.setJarByClass(GradeCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

This example outlines a basic Hadoop MapReduce job that counts occurrences of each grade, showing how to set up the mapper and reducer classes and configure the job.

These examples provide a starting point for students to explore and implement data processing tasks using Apache Spark and Hadoop in their academic projects.s students to focus on learning and applying their knowledge effectively.

Conclusion

Understanding the differences between Apache Spark and Hadoop is crucial for students embarking on big data projects. While Spark offers speed and ease of use, Hadoop provides robust storage and processing capabilities. Selecting the right framework depends on the project's specific needs, data size, and deadlines. By mastering both tools over time, students can position themselves for success in the ever-evolving field of big data.

Choosing the right framework is a step toward academic success, but students should continue to expand their skills and knowledge in both Spark and Hadoop to stay competitive in their future careers.

В прошлых статьях мы выяснили: если задача тяжелая и требует Java (Spark), мы используем SparkSubmitOperator. Но что делать, если у вас "тяжелый" Python? Типичная ситуация когда вы написали отличный код на Pandas внутри PythonOperator. На тестовом файле в 100 Мб все летало. В продакшене пришел файл на 10 Гб. Как результат OOM Kill (Out Of Memory). Воркер Airflow падает, задача фейлится, соседние легкие задачи тоже умирают, потому что процесс был убит операционной системой. Проблема архитектуры Airflow: PythonOperator выполняет код локально на том воркере, где он запущен. Это значит, что вы ограничены ресурсами одной машины. Пытаться наращивать RAM на воркере Airflow - это тупиковый путь. Решение: Нам нужно вынести исполнение кода за пределы Airflow, оставив за ним только функцию "кнопки пуск" и контроля статуса. Для Python-задач идеальным "внешним процессором" является Dask. В этой статье мы научим Airflow делегировать тяжелые вычисления удаленному кластеру, не меняя при этом привычный Python-стек.  

Архитектура делегирования - Как Airflow использует Dask для масштабирования

  В этой связке Airflow выступает в роли заказчика. Airflow Worker: Запускает задачу. Но вместо того чтобы грузить данные в свою память, он создает легкий объект-клиент. Сетевой вызов: Этот клиент стучится по TCP к планировщику Dask (Scheduler). Удаленное исполнение: Dask забирает инструкцию и данные (из S3), "перемалывает" их на своих мощностях. Ожидание: Airflow-задача висит в ожидании ответа (или мониторит статус), потребляя минимум ресурсов. Для Airflow это выглядит как обычный Python-скрипт, но физически нагрузка уходит на другие серверы.    

Настройка инфраструктуры для масштабирования выполнения задач Airflow

Чтобы Airflow мог управлять Dask-ом, внутри контейнера Airflow должна стоять библиотека dask, s3fs и pandas  нужных версий. Убедитесь, что она есть в вашем Dockerfile или установите её. обновляем dockerfile до: FROM apache/airflow:2.8.1-python3.10 USER root # 1. ОБЪЕДИНЕННАЯ УСТАНОВКА СИСТЕМНЫХ ПАКЕТОВ # gcc, libkrb5-dev... — нужны для компиляции HDFS провайдера (Урок 5) # default-jdk, procps, curl — нужны для работы Spark (Урок 6) RUN apt-get update && apt-get install -y --no-install-recommends gcc libkrb5-dev krb5-user libffi-dev default-jdk procps curl && apt-get autoremove -yqq --purge && apt-get clean && rm -rf /var/lib/apt/lists/* # 2. НАСТРОЙКА JAVA (Урок 6) ENV JAVA_HOME=/usr/lib/jvm/default-java # 3. УСТАНОВКА SPARK-КЛИЕНТА (Урок 6) ENV SPARK_VERSION=3.5.1 ENV HADOOP_VERSION=3 RUN curl -O https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && tar zxf spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz -C /opt/ && rm spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && ln -s /opt/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION} /opt/spark # 4. НАСТРОЙКА ПУТЕЙ SPARK ENV SPARK_HOME=/opt/spark ENV PATH=$PATH:$SPARK_HOME/bin: USER airflow #--- Добавляем ограничения на установку версий совместимых с Airflow и Python ARG AIRFLOW_VERSION=2.8.1 ARG PYTHON_VERSION=3.10 ARG CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt" # 5. УСТАНОВКА ПРОВАЙДЕРОВ # Ставим сразу оба: и для HDFS (чтобы работал DAG из 5 урока), и для Spark (Урок 6) #--- ОБНОВЛЕНИЕ ЗДЕСЬ --- RUN pip install --no-cache-dir "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}" apache-airflow-providers-apache-hdfs apache-airflow-providers-apache-spark "dask==2023.12.1" "distributed==2023.12.1" s3fs pandas # ПОСЛЕДНЯЯ СТРОКА Добавляем dask, distributed и s3fs (для работы с Yandex S3) только нужной версии Build-им снова docker image airflow-hdfs:2.8.1

Но и это еще не все к сожалению многокомпонентные распределенные системы, часто конфликтуют из-за  совместимости и на стороне выполнения задач ( Dask worker и scheduler). Dockerfile, который мы подготовили, настроен для сборки Airflow  контейнеров, и если вы соберете сейчас имидж с помощью ниже описанного docker-compose.yaml файла, вылезут несовместимости по python библиотекам используемым на Dask узлах поэтому сделаем небольшую сборку для Dask: #--- соберем кастомный имидж для установки Dask контейнеров с треьования по версионности компонент FROM daskdev/dask:2023.12.1 COPY requirements-dask.txt /tmp/requirements-dask.txt RUN pip install --no-cache-dir -r /tmp/requirements-dask.txt && python -c "import s3fs,fsspec,pandas; print('OK', s3fs.__version__, fsspec.__version__, pandas.__version__)" Создадим requirements-dask.txt для сборки нового имиджа под Dask #-- версии подходящие для dask:2023.12.1 fsspec==2023.12.1 s3fs==2023.12.1 aiobotocore==2.7.0 botocore==1.31.64 boto3==1.28.64 pandas Variables c именем 'yandex_creds' # В формате JSON: {"key": "...", "secret": "..."} # ИЛИ (для теста) впишите свои ключи ниже, если не хотите возиться с Variables # aws_key = "ВАШ_ACCESS_KEY" # aws_secret = "ВАШ_SECRET_KEY" # Чтобы не светить ключи в коде, попробуем достать из Environment (если прокинули в docker-compose) import os aws_key = os.getenv("AWS_ACCESS_KEY_ID", "ЗАМЕНИТЕ_НА_КЛЮЧ_ЕСЛИ_НЕТ_ENV") aws_secret = os.getenv("AWS_SECRET_ACCESS_KEY", "ЗАМЕНИТЕ_НА_СЕКРЕТ_ЕСЛИ_НЕТ_ENV") # 4. Отправляем задачу на кластер future = client.submit(heavy_processing_task, BUCKET_NAME, S3_FILE_PATTERN, aws_key, aws_secret) logging.info("Задача отправлена в Dask. Ждем...") # Ждем завершения wait(future) # Получаем результат (путь к файлу) try: result_path = future.result() logging.info(f"Успех! Данные сохранены в: {result_path}") except Exception as e: logging.error(f"Ошибка вычислений в Dask: {e}") raise e client.close() with DAG( dag_id="07.dask_yandex_processing", start_date=datetime(2023, 1, 1), schedule=None, catchup=False, tags= ) as dag: run_on_dask = PythonOperator( task_id="run_on_dask_cluster", python_callable=offload_to_dask ) Не забудьте прописать свои ключи в DAG файл ( конечно с точки зрения это не правильно, но наверняка вы не столкнетесь с кучей ошибок при попытке сохранить результаты из dask worker ов, а это уже совсем другая история которую мы расскажем вам на курсе  ниже по ссылке     А вот и наш результат

Хотя конечно мы все равно получили предупреждения о ращличиях в версиях компонент на Dask кластере, но не критичных для нашего DAG кода.  

Главные ошибки при интеграции Dask (Airflow-Specific)

При такой схеме работы инженеры часто наступают на специфические грабли синхронизации. Проблема синхронизации окружений (Dependency Hell) Airflow "сериализует" (упаковывает) вашу функцию и отправляет её в Dask. Сценарий: Вы используете в функции библиотеку Dask версии 2023.5.0, которая стоит на Airflow. А на Dask-воркере стоит Dask 2026.1.2(latest). Результат: Задача упадет с ошибкой десериализации. Правило Airflow: Образы Docker для Airflow Worker и Dask Worker должны иметь идентичный набор Python-библиотек. Если обновляете requirements.txt в Airflow - обновляйте и в Dask. Давайте поразмышляем  немного над этим. В нашем случае при старте нашего DAG мы сразу получаем ошибку и пытаемся разобраться в причинах

Присмотритесь над выделенным фрагментом - мы  увидим разные версии на клиенте и на сервере ( worker+scheduler) для dask и Python. вспоминаем опции которые мы прописали в dockerfile, когда создавали имидж для поднятия Airflow с библиотеками Dask и имиджа "FROM apache/airflow:2.8.1", но версии утилит Dask и Python, да и других утилит мы не выбирали. Без споров на будущее и с уверенностью что изменение конфигурации имиджа не разрушит совместимость с нашими прошлыми уроками (1-6) и установленными компонентами (Hadoop,Spark) выбираем фиксированную версию исходного имиджа Airflow c Python 3.10  

В dockerfile вносим изменения вместо "FROM apache/airflow:2.8.1" ->"FROM apache/airflow:2.8.1-python3.10" и для установки пакетов Dask, Distributed тоже жестко фиксируем номера версий ("dask==2023.12.1" "distributed==2023.12.1" "dask==2023.12.1" s3fs pandas), чтобы и в docker-compose.yaml при конфигурации dask workers и schedulers использовал не "image: daskdev/dask:latest", а версию максимально близкую и совместимую с клиентом допустим "image: daskdev/dask:2023.12.1".

Пересобираем docker image и проверяем работу DAGа  ( займет чуть больше времени) docker compose down docker build --no-cache -t airflow-hdfs:2.8.1 . docker compose up -d    Ошибка "Вернуть данные в return" Новички часто пишут: return heavy_dataframe. Что происходит: Dask возвращает гигабайты данных по сети обратно в Airflow Client. Airflow пытается записать это в XCom (мета-базу Postgres). Результат: База зависает, Airflow падает. Правило: Задачи на Dask должны читать из S3 и писать в S3. В Airflow возвращаем только пути к файлам или статус (True/False). Сетевая доступность Если Airflow запущен локально, а Dask в Docker (или наоборот), Client(...) не сможет подключиться. В нашем примере все работает, потому что оба сервиса живут в одной сети docker-compose. В реальном проде Airflow и Dask могут быть в разных подсетях Kubernetes, и вам придется настраивать доступы. Исправьте финальные варианты кода Dags  и  конфигурационных файлов и при необходимости сравните с нашими на GitHub где лежит код к Уроку 7.   Роль Cursor в написании "оберток" для масштабирование Airflow c Dask   Код делегирования часто шаблонный. Cursor может сэкономить время. Промпт: Напиши функцию-обертку для PythonOperator Airflow. Функция должна принимать адрес Dask-планировщика и словарь с параметрами. Внутри она должна подключаться к кластеру, запускать переданную функцию processing_logic, ждать результата и логировать прогресс. Обработай возможные ошибки подключения (TimeOut). Итог: Мы научились использовать PythonOperator не как исполнителя, а как контроллер. Теперь Airflow может запускать задачи любой тяжести, делегируя их Dask-кластеру. Мы решили проблему нехватки памяти (OOM) архитектурным способом, не меняя язык программирования. Теперь, когда мы умеем работать с тяжелыми файлами, пришло время поговорить о скорости реакции. В следующей статье мы разберем Event-Driven архитектуру. Мы узнаем, как заставить Airflow запускать DAG не по расписанию, а мгновенно - как только в Kafka прилетело сообщение о событии. Переходим к Kafka и событийной модели?      

Использованные референсы и материалы

Мы построили пайплайн, где данные забираются из базы и бережно складываются в HDFS. Теперь они лежат там мертвым грузом. Чтобы превратить сырые CSV в полезные отчеты, их нужно обработать: отфильтровать, агрегировать, джойнить. Делать это внутри самого Airflow (через PythonOperator и Pandas) - плохая идея если: Память: Если файл весит 100 ГБ, ваш воркер Airflow просто лопнет (OOM Kill). Скорость: Pandas работает на одном ядре. Spark распределяет задачу на сотни ядер. Airflow здесь выступает только как кнопка "Пуск". Он запускает задачу на кластере Spark и смиренно ждет, пока "большой брат" закончит работу. Для этого используется SparkSubmitOperator.  

Главная проблема - Airflow не умеет заходить в Spark "из коробки"

Здесь новички ломают копья. SparkSubmitOperator - это, по сути, обертка над консольной командой spark-submit. Чтобы этот оператор сработал, на машине (или в контейнере), где крутится воркер Airflow, должны быть физически установлены: Java (OpenJDK) — потому что Spark работает на JVM. Клиент Spark — набор бинарных файлов, чтобы отправить команду кластеру. Стандартный образ apache/airflow, который мы использовали до этого, не содержит Java. Если вы попробуете запустить Spark-задачу сейчас, вы получите ошибку JAVA_HOME not set или spark-submit not found. Помните как мы с вами костылизировали с Apache Hadoop и AirFlow на прошлом уроке?       Шаг 1. Модернизируем Docker-образ для поддержки Java и Apache Spark   Нам придется снова немного "запачкать руки" и пересоздать свой образ Airflow, который мы с вами использовали на прошлом уроке ( поддержка Hadoop provider) и добавить туда Spark с поддержкой Java. Не пугайтесь, это стандартная практика. Создайте(или отредактируйте существующий)  файл Dockerfile в корне проекта (рядом с docker-compose): FROM apache/airflow:2.8.1 USER root # 1. ОБЪЕДИНЕННАЯ УСТАНОВКА СИСТЕМНЫХ ПАКЕТОВ # gcc, libkrb5-dev... — нужны для компиляции HDFS провайдера (Урок 5) # default-jdk, procps, curl — нужны для работы Spark (Урок 6) RUN apt-get update && apt-get install -y --no-install-recommends gcc libkrb5-dev krb5-user libffi-dev default-jdk procps curl && apt-get autoremove -yqq --purge && apt-get clean && rm -rf /var/lib/apt/lists/* # 2. НАСТРОЙКА JAVA (Урок 6) ENV JAVA_HOME=/usr/lib/jvm/default-java export JAVA_HOME # 3. УСТАНОВКА SPARK-КЛИЕНТА (Урок 6) ENV SPARK_VERSION=3.5.1 ENV HADOOP_VERSION=3 RUN curl -O https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && tar zxf spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz -C /opt/ && rm spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION}.tgz && ln -s /opt/spark-${SPARK_VERSION}-bin-hadoop${HADOOP_VERSION} /opt/spark # 4. НАСТРОЙКА ПУТЕЙ SPARK ENV SPARK_HOME=/opt/spark ENV PATH=$PATH:$SPARK_HOME/bin USER airflow # 5. УСТАНОВКА ПРОВАЙДЕРОВ # Ставим сразу оба: и для HDFS (чтобы работал DAG из 5 урока), и для Spark RUN pip install --no-cache-dir apache-airflow-providers-apache-hdfs apache-airflow-providers-apache-spark Пересоздаем наш образ "франкенштейн" состоящий теперь из бинарников Hadoop, Spark и провайдеров для них в Airflow вместе с Java - такова цена интеграции. #--- и пересобираем его с нуля для уверенности - процесс может занять 5-10 минут docker build --no-cache -t airflow-hdfs:2.8.1 . #--- почистим старые images docker image ls #для проверки docker image prune docker image ls #для проверки #--- Уже запущенные контейнеры не обновят автоматически используемый image без опции --build docker compose up -d --build

  Теперь нужно сказать docker-compose, чтобы он использовал вновь созданный файл image 'airflow-hdfs:2.8.1', а не готовый образ из интернета. Измените секцию x-airflow-common в docker-compose.yaml добавив новые переменные и volumes для Spark Jobs: x-airflow-common: &airflow-common image: airflow-hdfs:2.8.1 # Изменяем используемый имидж # build: . environment: # ... старые переменные ... JAVA_HOME: /usr/lib/jvm/default-java # Добавляем новые переменные для Spark SPARK_HOME: /opt/spark volumes: # --- НОВЫЙ ТОМ ДЛЯ УРОКА 6 (Скрипты Spark) --- - ${AIRFLOW_PROJ_DIR:-.}/jobs:/opt/airflow/jobs   Шаг 2. Добавляем Spark-кластер   Нам нужен сам Spark, который будет выполнять работу. Добавим мастер и один воркер в docker-compose.yaml (в секцию services): # Урок 6 Spark jobs - кластер для Airflow tasks spark-master: image: apache/spark:3.5.1 container_name: spark-master hostname: spark-master profiles: - spark environment: - SPARK_NO_DAEMONIZE=true - SPARK_MASTER_HOST=spark-master ports: - "9090:8080" # Web UI (сместили на 9090, чтобы не конфликтовал с Airflow) - "7077:7077" # Мастер-порт command: /opt/spark/bin/spark-class org.apache.spark.deploy.master.Master spark-worker: image: apache/spark:3.5.1 container_name: spark-worker profiles: - spark environment: - SPARK_NO_DAEMONIZE=true - SPARK_WORKER_CORES=1 - SPARK_WORKER_MEMORY=1G depends_on: - spark-master command: /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker spark://spark-master:7077 Не забудьте включить hadoop & datanode из прошлого примера или применить --profile spark.  Теперь выполняем команду пересборки и запуска: docker-compose up -d --build

Зайдите на localhost:9090. Вы должны увидеть интерфейс Spark Master с одним живым воркером (Alive Workers: 1).

  Шаг 3. Настраиваем Connection Airflow должен знать, где живет Spark Master. Admin -> Connections. Новое соединение: Conn Id my_spark_conn Conn Type Spark Host spark://spark-master Port 7077 Extra ( или Deploy mode) {"deploy-mode": "client"} или cluster, но для Docker проще client, чтобы видеть логи сразу   Практика: PySpark Job для обработки данных   В прошлой статье мы положили файл users_{ds}.csv в HDFS. Давайте напишем скрипт на PySpark, который читает этот файл, считает распределение пользователей по датам и сохраняет отчет обратно в HDFS (или выводит в консоль). Создайте папку jobs рядом с dags и положите туда скрипт user_analytics.py: import sys from pyspark.sql import SparkSession from pyspark.sql.functions import col, count def main(input_path, output_path): # Создаем сессию spark = SparkSession.builder .appName("AirflowUserAnalytics") .getOrCreate() print(f"Reading from: {input_path}") # Читаем CSV. Так как хедер мы писали сами, указываем header=True df = spark.read.option("header", "true").csv(input_path) # Простая аналитика: сколько регистраций в каждую дату report = df.groupBy("date").agg(count("*").alias("total_users")) report.show() # Сохраняем результат (в формате Parquet, это стандарт для Big Data) # report.write.mode("overwrite").parquet(output_path) spark.stop() if __name__ == "__main__": # Аргументы передаются из Airflow if len(sys.argv) != 3: print("Usage: user_analytics.py ") sys.exit(-1) main(sys.argv, sys.argv)   Шаг 4. AirFlow DAG со SparkSubmitOperator    Теперь самое главное - связать всё вместе. Код DAG-а (dags/process_spark.py): from airflow import DAG from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator from datetime import datetime # Путь к HDFS, куда мы писали в прошлой статье # Обратите внимание: spark внутри Docker-сети может обращаться к namenode HDFS_INPUT = "hdfs://namenode:8020/user/airflow/backup/users_{{ ds }}.csv" HDFS_OUTPUT = "hdfs://namenode:8020/user/airflow/reports/daily_{{ ds }}" with DAG( dag_id="06.spark_processing", start_date=datetime(2023, 1, 1), schedule=None, catchup=False ) as dag: process_task = SparkSubmitOperator( task_id="run_spark_job", conn_id="my_spark_conn", # Путь к скрипту внутри контейнера AIRFLOW application="/opt/airflow/jobs/user_analytics.py", # Аргументы для скрипта (input и output) application_args=, # Конфигурация ресурсов (сколько отдать Спарку) conf={ "spark.driver.memory": "512m", "spark.executor.memory": "512m" }, # Важно! Указываем пакеты, если нужно работать с S3 или другими системами # packages="org.apache.hadoop:hadoop-aws:3.2.0", verbose=True ) Как это работает - объяснялки Airflow (Worker) берет параметры из оператора Он формирует длинную команду: /opt/spark/bin/spark-submit --master spark://spark-master:7077 ... user_analytics.py .... Эта команда запускается в контейнере Airflow Клиент Spark связывается с мастером, передает ему код Мастер распределяет задачу на spark-worker Воркер читает данные из HDFS, считает и пишет результат Airflow видит, что процесс spark-submit завершился с кодом 0, и красит задачу в зеленый цвет

  Давайте попробуем?  

Конечно проверить финальный результат из карманного Hadoop и в паркетном формате  тот еще дополнительный квест, но он того стоил ( через docker exec -it скопировать данные с hdfs на локальную систему namenode, потом через docker cp скопировать все с namenode локально и уже финально установив pandas и fastparquet прочитать все с использованием простого питон скрипта). По традиции все финальные версии файлов вы сможете найти у нас на Git    

Troubleshooting: Почему Spark падает?

Запуск Spark в контейнерах - это минное поле. Ошибка 1: JAVA_HOME is not set Причина: Вы не пересобрали Docker-образ или забыли ENV JAVA_HOME в Dockerfile. Лечение: Проверьте docker exec -it java -version. Если команды нет - пересобирайте образ. Ошибка 2: Connection refused к HDFS внутри Spark-джобы Причина: Spark Worker (отдельный контейнер) не может достучаться до Namenode. Лечение: Убедитесь, что все контейнеры (spark-worker, namenode) находятся в одной сети default (docker-compose делает это автоматически, но если вы запускали их разными файлами - будут проблемы). Ошибка 3: Driver runs, but Executor fails Причина: Часто это нехватка памяти. Spark по умолчанию может просить 1ГБ на экзекьютор, а выделили вы Докеру всего 2ГБ на всё. Лечение: Явно занижайте память в параметре conf: "spark.executor.memory": "512m".  

Помощь Cursor: Пишем PySpark код для DAG Airflow

  Если вы не помните синтаксис DataFrame API, Cursor сделает это за вас. "Напиши скрипт на PySpark, который читает Parquet-файлы из папки /data/input, фильтрует пользователей старше 18 лет, группирует их по городам и сортирует по убыванию количества. Результат сохрани в CSV."   Промпт для дебага: "Вот лог ошибки SparkSubmitOperator: . Объясни, почему Airflow не может найти класс org.apache.hadoop.fs.s3a.S3AFileSystem и какой пакет нужно добавить в spark-submit." Итог: Мы преодолели самый сложный барьер интеграции - настроили запуск Java-приложений из Python-оркестратора. Теперь Airflow управляет мощнейшим кластером обработки данных Apache Spark. Но Spark - это тяжело, долго и требует много памяти. Всегда ли нам нужен такой монстр если тем более у Вас его еще нет и не кому его приручить? В следующей статье мы рассмотрим легкую, питоническую альтернативу - Dask. Мы узнаем, как масштабировать Pandas-код на несколько серверов без установки Java и мучений с spark-submit. Переходим к Dask?    

Использованные референсы и материалы