Job Scheduling in Apache Spark

Job Scheduling in Apache Spark

Job Scheduling

Job scheduling có thể nằm ở 2 level: Cluster level hoặc Spark apps level

Cluster level: ở đây job sẽ là từng cái Spark apps, từng apps sẽ yêu cầu tài nguyên từ Cluster Manager cho việc xử lý job, khi yêu cầu đc chấp nhận, các tài nguyên này sẽ lock lại và k đc sử dụng bởi các Spark apps khác cho đến khi tài nguyên đc giải phóng

Application level: ở đây job sẽ là từng Action đc trigger trong 1 Spark apps. Sẽ có lúc có nhiều job được yêu cầu thực hiện cùng lúc, lúc này Job Scheduling sẽ sắp xếp thực hiện các job này 1 cách phù hợp ( FIFO, FAIR, … )

Job Scheduling at Cluster Level

Job Scheduling ở Cluster level có thể hiểu là Resource Availability của cụm. Khi 1 Spark apps được submit và chạy thì nó sẽ có các resource của riêng nó và các resource này sẽ ko share với các apps khác. Nếu có nhiều apps trên cùng 1 cluster cần share các resource với nhau thì chúng ta có 2 cách để phân bổ và quản lý resource:

Static Allocation ( phân bổ tĩnh )

Ở cách phân bổ này, 1 số lượng resource của cụm sẽ đc gán cứng cho 1 Spark apps trong lúc apps đang chạy. Các resource này sẽ đc giải phóng khi apps chạy xong.

Dynamic Allocation ( phân bổ động )

Ở cách phân bổ này, số lượng executor sẽ đc add và remove khỏi Spark apps theo nhu cầu, dựa vào lượng resource ước tính cần được sử dụng. Tính năng này sẽ rất có ích khi các Spark apps cần share lượng resource đang có trong cụm.

1 số config cơ bản

“spark.dynamicAllocation.enabled”, “spark.dynamicAllocation.minExecutors”, “spark.dynamicAllocation.maxExecutors”, “spark.dynamicAllocation.initialExecutors”.

config để khi executor bị remove, chúng sẽ ko xóa các file shuffle đã được ghi

“spark.shuffle.service.enabled”, “spark.dynamicAllocation.shuffleTracking.enabled”.

Add Request

“spark.dynamicAllocation.schedulerBacklogTimeout” , “spark.dynamicAllocation.sustainedSchedulerBacklogTimeout”.

Remove Request cho phép Spark apps remove 1 executor khi nó vượt quá số giây đc set bởi config dưới đây

“spark.dynamicAllocation.executorIdleTimeout”

khi executor đc add để thực thi task và sau đó remove, để bảo toàn đc cached data ta có thể nghiên cứu các config sau  “spark.shuffle.service.enabled”  “spark.dynamicAllocation.cachedExecutorIdleTimeout”, “spark.shuffle.service.fetch.rdd.enabled”

Job Scheduling at Application Level

FIFO ( Mặc định )

Nếu job đầu tiên ko dùng hết resource, job thứ 2 sẽ được start

Tuy nhiên nhược điểm là khi nếu bạn có 1 job đầu tiên chạy vài tiếng và job thứ 2 chỉ chạy vài giây thôi, nếu mà job đầu tiên lấy hết resource thì job thứ 2 sẽ phải đợi job đầu chạy xong mới đc chạy :((

⇒ Trường hợp này ta có thể dùng FAIR scheduling

FAIR ( sử dụng config “spark.scheduler.mode”: “FAIR” )

Fair scheduler sẽ phân bổ các task từ nhiều job khác nhau đến các con executor theo “round-robin fashion”. Cách scheduling có thể giúp job đc phân bổ resource đồng đều hơn.

Fair scheduler còn hỗ trợ chúng ta bỏ các job vào pool và phân bổ mức ưu tiên khác nhau đến các pool đó. Các pool sẽ được phân bổ resource dựa vào weight của chúng. Sử dụng pool có thể đảm bảo các job quan trọng sẽ được hoàn thành ( vì đc ưu tiên resource hơn )

Reference

https://medium.com/@badwaik.ojas/job-scheduling-in-apache-spark-e7b2ce2fbd

Comments

Post a Comment

Popular posts from this blog

Introduction to Apache Doris

Image
  Apache Doris là gì ? Apache Doris là một interactive SQL real-time data warehouse dựa vào kiến trúc MPP ( Massive Parallel Processing ). Hình ảnh dưới đây là những gì Apache Doris có thể đại diện trong data pipeline Technical Kiến trúc của Apache Doris được chia làm 2 loại Frontend (FE): user request processing: xử lý request từ người dùng query parsing and planning: xử lý parsing và xử lý plan của query metadata management: quản lý metadata node management: quản lý node Backend (BE): data storage: lưu trữ dữ liệu query execution: thực thi câu query Storage Engine: Apache Doris có engine lưu trữ dạng columnar ( encode, compress, read data by column ). Hỗ trợ nhiều cấu trúc index: Sorted Compound Key Index, MIN/MAX Indexing, Bloom Filter, Invert Index Hỗ trợ nhiều mô hình dữ liệu ( data model ): Aggregate Key Model, Unique Key Model, Duplicate Key Model Hỗ trợ Materialized Views ( tự động được select và update ) Query Engine: MPP-based query engine...
Read more

Một số câu hỏi về 'difference between'

Image
 Một số câu hỏi về 'difference between' ( P1 ) 1. What is the difference between row_number() and rank() in SQL ? Khác nhau ở chỗ xử lý giá trị bằng nhau, đối với: row_number(): thì sẽ đánh số tăng dần ( mặc định bắt đầu từ 1 ) đối với mỗi dòng rank(): thì sẽ đánh số có giá trị bằng nhau ( ví dụ 2 dòng đều là rank 1 ) đối với 2 dòng có giá trị bằng nhau 2. What is the difference between IN and EXISTS and providing me example ? Khác nhau ở chỗ khi filter các giá trị trong 1 list, đối với: in(): thì đó là 1 list biết trước các giá trị exists() thì đó là 1 list chưa biết trước giá trị ( ta sẽ chạy câu này để kiểm trả xem có match giá trị nằm trong vế subquery của exists hay ko ) 3. What is the difference between true column store and row store ? Khác nhau ở chỗ cách lưu trữ, đối với: row store: dữ liệu được nhóm với nhau theo hàng Row 1: [Value1, Value2, Value3] Row 2: [Value4, Value5, Value6] Row 3: [Value7, Value8, Value9] column store: dữ liệu được nhóm với nhau theo ...
Read more

Introduction to Apache Hudi

  Introduction to Apache Hudi Data Lake Integration Hudi có thể đẩy dữ liệu realtime vào các hệ lưu trữ distributed file systems (DFS), như là  HDFS  hoặc cloud stores ( MinIO ) Có thể chứa đc dữ liệu lớn trên file ( parquet, … ), tiết kiệm hơn so với lưu trữ trên các hệ OLAP ( Kudu, Pinot ) Tương thích với nhiều loại query engine nên ta có thể dễ dàng query data từ Hudi table Core Components Tables : Hudi tổ chức data dưới dạng bảng, dễ dàng quản lý và query. Transactions : Hudi hỗ trợ những hoạt động transactional như UPDATE , DELETE ( nếu chỉ dùng Hive thì ko có ) trên từng dòng data. Bảo đảm consistency và durability trong các hành động write data ( Tính chất C và D trong ACID ). Upserts/Deletes : Hudi hỗ trợ update và delete, cho phép bạn thay đổi dòng data hiện tại nhưng ko cần phải xử lý hết toàn bộ dataset ( toàn bộ file ) Advanced Indexes : Hudi hỗ trợ đánh index, giúp cải thiện query performance. Streaming Ingestion Services : Hudi hỗ trợ ing...
Read more