এয়ারফ্লো হল ব্যাচ ডেটা প্রসেসিং প্রক্রিয়াগুলিকে সুবিধামত এবং দ্রুত বিকাশ ও বজায় রাখার একটি হাতিয়ার

হ্যালো, হাবর! এই নিবন্ধে আমি ব্যাচ ডেটা প্রসেসিং প্রক্রিয়াগুলি বিকাশের জন্য একটি দুর্দান্ত সরঞ্জাম সম্পর্কে কথা বলতে চাই, উদাহরণস্বরূপ, একটি কর্পোরেট DWH বা আপনার DataLake এর পরিকাঠামোতে। আমরা Apache Airflow সম্পর্কে কথা বলব (এখন থেকে Airflow হিসাবে উল্লেখ করা হয়েছে)। এটি অন্যায়ভাবে Habré-এ মনোযোগ থেকে বঞ্চিত, এবং মূল অংশে আমি আপনাকে বোঝানোর চেষ্টা করব যে আপনার ETL/ELT প্রক্রিয়াগুলির জন্য একটি সময়সূচী নির্বাচন করার সময় অন্তত এয়ারফ্লোটি দেখার মূল্য।

পূর্বে, আমি যখন Tinkoff ব্যাংকে কাজ করতাম তখন আমি DWH বিষয়ের উপর একটি সিরিজ নিবন্ধ লিখেছিলাম। এখন আমি Mail.Ru গ্রুপ টিমের অংশ হয়েছি এবং গেমিং এলাকায় ডেটা বিশ্লেষণের জন্য একটি প্ল্যাটফর্ম তৈরি করছি। প্রকৃতপক্ষে, সংবাদ এবং আকর্ষণীয় সমাধানগুলি উপস্থিত হওয়ার সাথে সাথে আমার দল এবং আমি এখানে ডেটা বিশ্লেষণের জন্য আমাদের প্ল্যাটফর্ম সম্পর্কে কথা বলব।

প্রস্তাবনা

তো, শুরু করা যাক। বায়ুপ্রবাহ কি? এটি একটি লাইব্রেরি (বা লাইব্রেরি সেট) কাজের প্রক্রিয়াগুলি বিকাশ, পরিকল্পনা এবং নিরীক্ষণ করা। বায়ুপ্রবাহের প্রধান বৈশিষ্ট্য: পাইথন কোডটি প্রক্রিয়াগুলি বর্ণনা (বিকাশ) করতে ব্যবহৃত হয়। আপনার প্রজেক্ট এবং ডেভেলপমেন্ট সংগঠিত করার জন্য এটির অনেক সুবিধা রয়েছে: সংক্ষেপে, আপনার (উদাহরণস্বরূপ) ETL প্রোজেক্টটি শুধুমাত্র একটি পাইথন প্রজেক্ট, এবং আপনি এটিকে আপনার ইচ্ছামতো সংগঠিত করতে পারেন, পরিকাঠামো, দলের আকার এবং বিশেষত্ব বিবেচনা করে। অন্যান্য প্রয়োজনীয়তা. যন্ত্রগতভাবে সবকিছু সহজ। উদাহরণস্বরূপ PyCharm + Git ব্যবহার করুন। এটা বিস্ময়কর এবং খুব সুবিধাজনক!

এখন আসুন বায়ুপ্রবাহের প্রধান সত্তাগুলি দেখি। তাদের সারমর্ম এবং উদ্দেশ্য বোঝার দ্বারা, আপনি সর্বোত্তমভাবে আপনার প্রক্রিয়া আর্কিটেকচার সংগঠিত করতে পারেন। সম্ভবত প্রধান সত্তা হল নির্দেশিত অ্যাসাইক্লিক গ্রাফ (এর পরে DAG হিসাবে উল্লেখ করা হয়েছে)।

DAG

একটি DAG হল আপনার কাজের কিছু অর্থপূর্ণ সংঘ যা আপনি একটি নির্দিষ্ট সময়সূচী অনুযায়ী কঠোরভাবে সংজ্ঞায়িত ক্রমানুসারে সম্পন্ন করতে চান। এয়ারফ্লো DAG এবং অন্যান্য সত্তার সাথে কাজ করার জন্য একটি সুবিধাজনক ওয়েব ইন্টারফেস প্রদান করে:

DAG এর মত দেখতে পারে:

ডেভেলপার, একটি DAG ডিজাইন করার সময়, অপারেটরদের একটি সেট লেখেন যার উপর DAG-এর মধ্যে কাজগুলি তৈরি করা হবে। এখানে আমরা আরেকটি গুরুত্বপূর্ণ সত্তায় আসি: এয়ারফ্লো অপারেটর।

অপারেটার

একটি অপারেটর হল এমন একটি সত্তা যার ভিত্তিতে কাজের উদাহরণ তৈরি করা হয়, যা একটি কাজের উদাহরণ কার্যকর করার সময় কী ঘটবে তা বর্ণনা করে। গিটহাব থেকে এয়ারফ্লো রিলিজ হয় ইতিমধ্যে ব্যবহার করার জন্য প্রস্তুত অপারেটর একটি সেট রয়েছে. উদাহরণ:

• BashOperator - একটি bash কমান্ড কার্যকর করার জন্য অপারেটর।
• PythonOperator - পাইথন কোড কল করার জন্য অপারেটর।
• ইমেইল অপারেটর — ইমেল পাঠানোর জন্য অপারেটর।
• HTTPOperator - HTTP অনুরোধের সাথে কাজ করার জন্য অপারেটর।
• SqlOperator - SQL কোড নির্বাহের জন্য অপারেটর।
• সেন্সর হল একটি ইভেন্টের জন্য অপেক্ষা করার জন্য একটি অপারেটর (প্রয়োজনীয় সময়ের আগমন, প্রয়োজনীয় ফাইলের উপস্থিতি, ডাটাবেসের একটি লাইন, API থেকে একটি প্রতিক্রিয়া, ইত্যাদি)।

আরো নির্দিষ্ট অপারেটর আছে: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator।

আপনি আপনার নিজস্ব বৈশিষ্ট্যের উপর ভিত্তি করে অপারেটর তৈরি করতে পারেন এবং আপনার প্রকল্পে তাদের ব্যবহার করতে পারেন। উদাহরণস্বরূপ, আমরা MongoDBToHiveViaHdfsTransfer, MongoDB থেকে Hive-এ নথি রপ্তানির জন্য একটি অপারেটর এবং কাজ করার জন্য বেশ কয়েকটি অপারেটর তৈরি করেছি ক্লিকহাউস: CHLoadFromHiveOperator এবং CHTableLoaderOperator. মূলত, যত তাড়াতাড়ি একটি প্রকল্প ঘন ঘন মৌলিক বিবৃতিতে নির্মিত কোড ব্যবহার করে, আপনি এটি একটি নতুন বিবৃতিতে তৈরি করার কথা ভাবতে পারেন। এটি আরও উন্নয়নকে সহজ করবে, এবং আপনি প্রকল্পে আপনার অপারেটরদের লাইব্রেরি প্রসারিত করবেন।

এর পরে, এই সমস্ত কাজের দৃষ্টান্তগুলি সম্পাদন করা দরকার, এবং এখন আমরা সময়সূচী সম্পর্কে কথা বলব।

সময়সূচী

এয়ারফ্লো এর টাস্ক শিডিউলার তৈরি করা হয়েছে সেলারি. সেলেরি হল একটি পাইথন লাইব্রেরি যা আপনাকে একটি সারি এবং অ্যাসিঙ্ক্রোনাস এবং ডিস্ট্রিবিউটেড কার্য সম্পাদন করার অনুমতি দেয়। বায়ুপ্রবাহের দিকে, সমস্ত কাজ পুলগুলিতে বিভক্ত। পুলগুলি ম্যানুয়ালি তৈরি করা হয়। সাধারণত, তাদের উদ্দেশ্য উত্সের সাথে কাজ করার কাজের চাপকে সীমিত করা বা DWH-এর মধ্যে কাজগুলি টাইপ করা। পুলগুলি ওয়েব ইন্টারফেসের মাধ্যমে পরিচালনা করা যেতে পারে:

প্রতিটি পুলের স্লটের সংখ্যার একটি সীমা রয়েছে। একটি DAG তৈরি করার সময়, এটি একটি পুল দেওয়া হয়:

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

DAG স্তরে সংজ্ঞায়িত একটি পুল টাস্ক লেভেলে ওভাররাইড করা যেতে পারে।
একটি পৃথক প্রক্রিয়া, শিডিউলার, এয়ারফ্লোতে সমস্ত কাজের সময় নির্ধারণের জন্য দায়ী। প্রকৃতপক্ষে, নির্ধারক কার্য সম্পাদনের জন্য সেট করার সমস্ত মেকানিক্স নিয়ে কাজ করে। কাজটি কার্যকর হওয়ার আগে বেশ কয়েকটি ধাপ অতিক্রম করে:

1. ডিএজি-তে আগের কাজগুলো সম্পন্ন হয়েছে; নতুন একটি সারিবদ্ধ করা যেতে পারে।
2. সারিটি কাজের অগ্রাধিকারের উপর নির্ভর করে সাজানো হয় (অগ্রাধিকারগুলিও নিয়ন্ত্রণ করা যেতে পারে), এবং যদি পুলে একটি বিনামূল্যের স্লট থাকে তবে কাজটি কার্যকর করা যেতে পারে।
3. যদি একটি বিনামূল্যে কর্মী সেলারি আছে, টাস্ক এটি পাঠানো হয়; সমস্যাটিতে আপনি যে কাজটি প্রোগ্রাম করেছেন তা এক বা অন্য অপারেটর ব্যবহার করে শুরু হয়।

যথেষ্ট সহজ।

শিডিউলার সমস্ত DAG-এর সেটে এবং DAG-এর মধ্যে সমস্ত কাজ চালায়।

DAG-এর সাথে কাজ শুরু করার জন্য, DAG-কে একটি সময়সূচী সেট করতে হবে:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

রেডিমেড প্রিসেটগুলির একটি সেট রয়েছে: @once, @hourly, @daily, @weekly, @monthly, @yearly.

আপনি ক্রন এক্সপ্রেশন ব্যবহার করতে পারেন:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

মৃত্যুদন্ড তারিখ

এয়ারফ্লো কীভাবে কাজ করে তা বোঝার জন্য, DAG-এর জন্য এক্সিকিউশন ডেট কী তা বোঝা গুরুত্বপূর্ণ। এয়ারফ্লোতে, DAG-এর একটি এক্সিকিউশন ডেট ডাইমেনশন থাকে, অর্থাৎ, DAG-এর কাজের সময়সূচীর উপর নির্ভর করে, প্রতিটি এক্সিকিউশন ডেটের জন্য টাস্ক ইনস্ট্যান্স তৈরি করা হয়। এবং প্রতিটি এক্সিকিউশন তারিখের জন্য, টাস্কগুলি পুনরায় কার্যকর করা যেতে পারে - বা, উদাহরণস্বরূপ, একটি ডিএজি একাধিক এক্সিকিউশন তারিখে একই সাথে কাজ করতে পারে। এটি এখানে স্পষ্টভাবে দেখানো হয়েছে:

দুর্ভাগ্যবশত (বা সৌভাগ্যবশত: এটি পরিস্থিতির উপর নির্ভর করে), যদি ডিএজি-তে টাস্কের বাস্তবায়ন সংশোধন করা হয়, তাহলে পূর্ববর্তী এক্সিকিউশন ডেটে এক্সিকিউশনটি সামঞ্জস্য বিবেচনায় নিয়ে এগিয়ে যাবে। আপনার যদি একটি নতুন অ্যালগরিদম ব্যবহার করে বিগত সময়ের ডেটা পুনরায় গণনা করার প্রয়োজন হয় তবে এটি খারাপ, তবে এটি খারাপ কারণ ফলাফলের পুনরুত্পাদনযোগ্যতা হারিয়ে গেছে (অবশ্যই, গিট থেকে সোর্স কোডের প্রয়োজনীয় সংস্করণটি ফেরত দিতে এবং কী গণনা করতে আপনাকে কেউ বিরক্ত করে না। আপনার একবার প্রয়োজন, যেভাবে আপনার প্রয়োজন)।

কাজ তৈরি করা হচ্ছে

DAG এর বাস্তবায়ন হল পাইথনে কোড, তাই কাজ করার সময় কোডের পরিমাণ কমানোর জন্য আমাদের কাছে একটি খুব সুবিধাজনক উপায় রয়েছে, উদাহরণস্বরূপ, শার্ড সোর্স সহ। ধরা যাক আপনার কাছে একটি উৎস হিসাবে তিনটি MySQL শার্ড রয়েছে, আপনাকে প্রতিটিতে আরোহণ করতে হবে এবং কিছু ডেটা তুলতে হবে। তাছাড়া, স্বাধীনভাবে এবং সমান্তরালভাবে। DAG-তে পাইথন কোডটি এইরকম দেখতে পারে:

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

DAG এই মত দেখায়:

এই ক্ষেত্রে, আপনি কেবল সেটিংস সামঞ্জস্য করে এবং DAG আপডেট করে একটি শার্ড যোগ করতে বা সরাতে পারেন। আরামপ্রদ!

আপনি আরও জটিল কোড জেনারেশন ব্যবহার করতে পারেন, উদাহরণস্বরূপ, একটি ডাটাবেসের আকারে উত্সগুলির সাথে কাজ করুন বা একটি টেবিলের কাঠামো বর্ণনা করুন, একটি টেবিলের সাথে কাজ করার জন্য একটি অ্যালগরিদম এবং, DWH পরিকাঠামোর বৈশিষ্ট্যগুলি বিবেচনায় নিয়ে একটি প্রক্রিয়া তৈরি করুন আপনার স্টোরেজে N টেবিল লোড করার জন্য। অথবা, উদাহরণস্বরূপ, একটি API এর সাথে কাজ করা যা একটি তালিকা আকারে একটি প্যারামিটারের সাথে কাজ করা সমর্থন করে না, আপনি এই তালিকা থেকে একটি DAG-তে N কাজ তৈরি করতে পারেন, API-এ অনুরোধের সমান্তরালতাকে একটি পুলে সীমাবদ্ধ করতে পারেন এবং স্ক্র্যাপ করতে পারেন। API থেকে প্রয়োজনীয় তথ্য। নমনীয় !

ভান্ডার

এয়ারফ্লো এর নিজস্ব ব্যাকএন্ড রিপোজিটরি আছে, একটি ডাটাবেস (মাইএসকিউএল বা পোস্টগ্রেস হতে পারে, আমাদের কাছে পোস্টগ্রেস রয়েছে), যা কাজগুলির অবস্থা, ডিএজি, সংযোগ সেটিংস, গ্লোবাল ভেরিয়েবল ইত্যাদি সংরক্ষণ করে। এখানে আমি বলতে চাই যে এয়ারফ্লোতে সংগ্রহস্থল খুবই সহজ (প্রায় 20টি টেবিল) এবং সুবিধাজনক যদি আপনি এটির উপরে আপনার নিজস্ব কোনো প্রক্রিয়া তৈরি করতে চান। আমি ইনফরম্যাটিকা রিপোজিটরিতে 100500 টেবিলের কথা মনে রাখি, যেগুলি কীভাবে একটি প্রশ্ন তৈরি করতে হয় তা বোঝার আগে দীর্ঘ সময়ের জন্য অধ্যয়ন করতে হয়েছিল।

পর্যবেক্ষণ

সংগ্রহস্থলের সরলতা দেওয়া, আপনি একটি টাস্ক পর্যবেক্ষণ প্রক্রিয়া তৈরি করতে পারেন যা আপনার জন্য সুবিধাজনক। আমরা জেপেলিনে একটি নোটপ্যাড ব্যবহার করি, যেখানে আমরা কাজের স্থিতি দেখি:

এটি Airflow এর ওয়েব ইন্টারফেসও হতে পারে:

এয়ারফ্লো কোডটি ওপেন সোর্স, তাই আমরা টেলিগ্রামে সতর্কতা যোগ করেছি। একটি টাস্কের প্রতিটি চলমান উদাহরণ, যদি একটি ত্রুটি ঘটে, তাহলে টেলিগ্রামে গ্রুপটিকে স্প্যাম করে, যেখানে সমগ্র উন্নয়ন এবং সহায়তা দল থাকে।

আমরা টেলিগ্রাম (যদি প্রয়োজন হয়) এর মাধ্যমে একটি তাত্ক্ষণিক প্রতিক্রিয়া পাই এবং জেপেলিনের মাধ্যমে আমরা এয়ারফ্লোতে কাজের একটি সামগ্রিক চিত্র পাই।

মোটে

বায়ুপ্রবাহ প্রাথমিকভাবে ওপেন সোর্স, এবং আপনার এটি থেকে অলৌকিক ঘটনা আশা করা উচিত নয়। কাজ করে এমন একটি সমাধান তৈরি করতে সময় এবং প্রচেষ্টা দেওয়ার জন্য প্রস্তুত থাকুন। লক্ষ্য অর্জনযোগ্য, বিশ্বাস করুন, এটি মূল্যবান। বিকাশের গতি, নমনীয়তা, নতুন প্রক্রিয়া যুক্ত করার সহজতা - আপনি এটি পছন্দ করবেন। অবশ্যই, আপনাকে প্রকল্পের সংগঠনের দিকে অনেক মনোযোগ দিতে হবে, বায়ুপ্রবাহের স্থিতিশীলতা: অলৌকিক ঘটনা ঘটবে না।

এখন আমাদের এয়ারফ্লো প্রতিদিন কাজ করে প্রায় 6,5 হাজার কাজ. তারা চরিত্রে বেশ ভিন্ন। অনেকগুলি বিভিন্ন এবং খুব নির্দিষ্ট উত্স থেকে মূল DWH-এ ডেটা লোড করার কাজ রয়েছে, প্রধান DWH-এর ভিতরে স্টোরফ্রন্ট গণনা করার কাজ রয়েছে, একটি দ্রুত DWH-এ ডেটা প্রকাশ করার কাজ রয়েছে, অনেকগুলি বিভিন্ন কাজ রয়েছে - এবং বায়ুপ্রবাহ দিনের পর দিন তাদের সব চিবিয়ে. সংখ্যায় কথা বলছি, এই 2,3 হাজার DWH (Hadoop) এর মধ্যে বিভিন্ন জটিলতার ELT কাজ, প্রায়। 2,5 শত ডাটাবেস সূত্র, এটি থেকে একটি দল 4 ETL ডেভেলপার, যা DWH-এ ETL ডেটা প্রসেসিং এবং DWH-এর ভিতরে ELT ডেটা প্রসেসিং-এ বিভক্ত এবং অবশ্যই আরও একজন অ্যাডমিন, যারা পরিষেবার পরিকাঠামো নিয়ে কাজ করে।

ভবিষ্যতের জন্য পরিকল্পনা

প্রক্রিয়ার সংখ্যা অনিবার্যভাবে বাড়ছে, এবং বায়ুপ্রবাহের পরিকাঠামোর ক্ষেত্রে আমরা যে প্রধান কাজটি করব তা হল স্কেলিং। আমরা একটি এয়ারফ্লো ক্লাস্টার তৈরি করতে চাই, সেলারি কর্মীদের জন্য একজোড়া পা বরাদ্দ করতে চাই, এবং কাজের সময় নির্ধারণের প্রক্রিয়া এবং একটি সংগ্রহস্থল সহ একটি স্ব-নকল মাথা তৈরি করতে চাই৷

উপসংহার

এটি অবশ্যই, আমি এয়ারফ্লো সম্পর্কে বলতে চাই এমন সবকিছু নয়, তবে আমি মূল পয়েন্টগুলি হাইলাইট করার চেষ্টা করেছি। খাওয়ার সাথে ক্ষুধা আসে, চেষ্টা করে দেখুন আপনার ভালো লাগবে :)

উত্স: www.habr.com