Кеңейтілген Spark тапсырмаларымен жұмыс істеу

Spark тапсырмаларын кластерге әртүрлі тәсілдермен жіберуге болады:

• Тәуелділіктері жоқ Spark қолданбалары үшін қолданба кодын тапсырма манифесіне беру жеткілікті. Бұл тәсіл π санын есептеу мысалында көрсетілген.
• Өз жұмысы үшін қосымша артефактілерді қажет ететін Spark қолданбалары үшін қажетті артефактілерді VK Object Storage бакетіне қолмен қосып, тапсырма манифесін өңдеу қажет. Бұл тәсіл төменде көрсетіледі.

Мысал ретінде Cloud Databases инстансы түрінде жайылған ClickHouse-қа SQL сұрауын орындайтын қолданба пайдаланылады.

1. Кітапхананы орнатыңыз, егер бұл әлі жасалмаған болса.

2. Қолжетімділік токенін жасаңыз Администратор рөлімен, егер бұл әлі жасалмаған болса. Бұл токен кітапханамен жұмыс істеу үшін қажет.

3. Cloud Spark кластерін жасаңыз.

   Кластер параметрлерін өз қалауыңыз бойынша таңдаңыз.

4. Скриптті орындап, жобадағы Spark кластерлері туралы ақпаратты алыңыз:

    
   from mlplatform_client import MLPlatformREFRESH_TOKEN = '<значение токена доступа>'mlp = MLPlatform(REFRESH_TOKEN)print(mlp.get_clusters())

   Кластерлер туралы толық ақпарат шығарылады.

5. Жасалған кластердің идентификаторын тауып, жазып алыңыз (id өрісінде болады).

6. ClickHouse инстансын жасаңыз, ол:

   • қолжетімді нұсқалардың ішіндегі ең жаңасын пайдаланады;
   • Single конфигурациясын пайдаланады;
   • сыртқы IP-мекенжайы бар және интернеттен қолжетімді;
   • user_spark пайдаланушысы бапталған;
   • db_spark жаңа дерекқорымен жұмыс істеуге бапталған.

   Пайдаланушының құпиясөзін жазып алыңыз, ол кейінірек қажет болады.

7. Жасалған ClickHouse инстансына тағайындалған IP-мекенжайды біліңіз:

   1. VK Cloud жеке кабинетіне өтіңіз.
   2. ДҚ инстансы орналасқан жобаны таңдаңыз.
   3. Базы данных → Инстансы баз данных бөліміне өтіңіз.
   4. Инстанс атауын басып, Информация қойындысына өтіңіз.
   5. Инстанстың сыртқы IP-мекенжайын жазып алыңыз.

Келесі мазмұндағы query-clickhouse.py файлын жасаңыз:

 
import osfrom pyspark.sql import SparkSessionCLICKHOUSE_IP = os.environ.get('CH_IP')CLICKHOUSE_PORT = 8123CLICKHOUSE_DB = 'db_spark'CLICKHOUSE_USER = os.environ.get('CH_USER')CLICKHOUSE_USER_PASSWORD = os.environ.get('CH_PASSWORD')CLICKHOUSE_JDBC_DRIVER = os.environ.get('CH_DRIVER_JAR')spark = SparkSession \    .builder \    .appName('PythonSparkClickHouse') \    .config('spark.driver.extraClassPath', f'./{CLICKHOUSE_JDBC_DRIVER}') \    .getOrCreate()print("Reading ClickHouse table")data_frame = spark.read.format('jdbc') \    .option('url', f'jdbc:clickhouse://{CLICKHOUSE_IP}:{CLICKHOUSE_PORT}/{CLICKHOUSE_DB}') \    .option('user', CLICKHOUSE_USER) \    .option('password', CLICKHOUSE_USER_PASSWORD) \    .option('query', 'SELECT * FROM system.build_options') \    .option('driver', 'com.clickhouse.jdbc.ClickHouseDriver') \    .load()data_frame.show();print("Finished reading ClickHouse table")spark.stop()

Бұл Spark қолданбасының коды, ол ClickHouse инстансына қосылып, SQL сұрауының көмегімен ClickHouse құрастырылымы (build) туралы ақпаратты сұратады.

Бұл қолданбаны пайдалануды жеңілдету үшін кейбір қосылу параметрлері орта айнымалылары арқылы беріледі:

• ClickHouse-қа қосылу CH_IP IP-мекенжайы бойынша CH_DRIVER_JAR JAR-файлында сақталған драйвердің көмегімен орындалады. Драйвер файлы кейінірек жүктеледі.
• ClickHouse-та аутентификация CH_USER және CH_PASSWORD деректемелерінің көмегімен орындалады. Мұндай тәсіл сезімтал деректерді тікелей кодқа орналастырмауға мүмкіндік береді.

Аталған орта айнымалыларының барлығы Spark тапсырмасын кластерге жіберу кезінде кейінірек орнатылады.

1. Кластер пайдаланатын VK Object Storage бакетінің атын анықтаңыз:

   1. Скриптті орындап, жобадағы Spark кластерлері туралы ақпаратты алыңыз:

       
      from mlplatform_client import MLPlatformREFRESH_TOKEN = '<значение токена доступа>'mlp = MLPlatform(REFRESH_TOKEN)print(mlp.get_clusters())

      Кластерлер туралы толық ақпарат шығарылады.

   2. Қажетті ақпаратты табыңыз:

      • кластер идентификаторы (id өрісінде болады);
      • VK Object Storage бакетінің атауы (s3_bucket_name өрісінде болады).

2. Файлдарды осы бакеттің spark-files директориясына жүктеңіз:

   • Spark қолданбасының коды бар query-clickhouse.py.

   • ClickHouse үшін JDBC драйвері бар clickhouse-jdbc-0.5.0-shaded.jar.

     Сілтемеде драйвер үшін барлық қажетті тәуелділіктерді қамтитын JAR-файлдың shaded-нұсқасы орналастырылған. Бұл драйверді Spark қолданбасымен пайдалануды және Spark тапсырмасын жіберу кезінде JAR-файлдармен жұмысты жеңілдетеді.

     Қажет болса, драйвердің басқа нұсқасын Maven репозиторийінен жүктеңіз. Бұл жағдайда төменде келтірілген скриптте драйвер нұсқасын қажеттісіне ауыстырыңыз.

Spark тапсырмасын жіберу кезінде бакетке жүктелген файлдарға жолдар көрсетіледі. Cloud Spark кластерінде бұл бакеттегі объектілерге қолжетімділік бұрыннан бар, қосымша баптаулар қажет емес.

1. Келесі мазмұндағы clickhouse-secret.yaml файлын жасаңыз:

    
   apiVersion: v1kind: Secretmetadata:  name: clickhouse-secretstringData:  CH_USER: user_spark  CH_PASSWORD: <пароль пользователя user_spark в ClickHouse>

   Бұл файл ClickHouse инстансына қосылуға арналған пайдаланушы деректемелері сақталатын Kubernetes құпиясын сипаттайды.

2. Скриптті орындап, тапсырманы кластерге жіберіңіз:

    
   from mlplatform_client import MLPlatformREFRESH_TOKEN = '<значение токена доступа>'CLUSTER_ID = '<идентификатор кластера>'BUCKET_NAME = '<имя бакета, который используется кластером>'CLICKHOUSE_IP = '<внешний IP-адрес инстанса ClickHouse>'CLICKHOUSE_JDBC_DRIVER = 'clickhouse-jdbc-0.5.0-shaded.jar'JOB_NAME = 'clickhouse-spark-job'mlp = MLPlatform(refresh_token=REFRESH_TOKEN)spark_job_manifest = mlp.get_default_manifest(cluster_id=CLUSTER_ID, job_name=JOB_NAME)spark_job_manifest.main_app_file = f's3a://{BUCKET_NAME}/spark-files/query-clickhouse.py'spark_job_manifest.add_jars([f's3a://{BUCKET_NAME}/spark-files/{CLICKHOUSE_JDBC_DRIVER}'])spark_job_manifest.add_driver_env(    [        {'name': 'CH_IP', 'value': CLICKHOUSE_IP},        {'name': 'CH_DRIVER_JAR', 'value': CLICKHOUSE_JDBC_DRIVER}    ])spark_job_manifest.add_executor_env(    [        {'name': 'CH_IP', 'value': CLICKHOUSE_IP},        {'name': 'CH_DRIVER_JAR', 'value': CLICKHOUSE_JDBC_DRIVER}    ])mlp.create_secret_from_yaml(cluster_id=CLUSTER_ID, secret_yaml_path='clickhouse-secret.yaml')spark_job_manifest.add_driver_env_from([{"secretRef": {"name": "clickhouse-secret"}}])spark_job_manifest.add_executor_env_from([{"secretRef": {"name": "clickhouse-secret"}}])spark_job_info = mlp.spark_submit_job(cluster_id=CLUSTER_ID, manifest=spark_job_manifest)print(spark_job_info)

   Жіберілген тапсырма туралы ақпарат шығарылады, мысалы:

    
   Job: clickhouse-spark-job, status: SUBMITTED, created_at: ...

   Бұған дейін кластер пайдаланатын бакетке Spark қолданбасының жұмысына қажетті файлдар жүктелген болатын. Әдепкі бойынша Spark тапсырмасына арналған манифестте қажетті файлдардың қайда орналасқаны туралы мәліметтер жоқ. Сондықтан манифестке жетіспейтін ақпарат қосылды:

   • spark_job_manifest.main_app_file ішіндегі Spark қолданбасының орындалатын файлына жол;
   • Қолданбаға қажет JAR-файлдарға жолдар тізімі (бұл жағдайда JDBC драйвері бар бір файл қажет). Бұл файлдарға жолдарды қосу үшін spark_job_manifest.add_jars() функциясы пайдаланылады.

   Сондай-ақ манифестке қажетті орта айнымалыларының мәндері қосылды. CH_USER және CH_PASSWORD айнымалылары ClickHouse-та аутентификация үшін пайдаланылады, сондықтан олардың мәндері clickhouse-secret Kubernetes құпиясынан алынады. Бұл құпия clickhouse-secret.yaml файлынан алдын ала жасалған.

1. Тапсырма логтарында ClickHouse-қа SQL сұрауын орындау нәтижесі пайда болғанына көз жеткізіңіз. Егер нәтиже шықпаса, логтарды алу үшін скриптті тағы бір рет іске қосыңыз: тапсырма әлі аяқталмаған сәттегі аралық жұмыс логтары шығарылуы мүмкін.

    
   from mlplatform_client import MLPlatformREFRESH_TOKEN = '<значение токена доступа>'CLUSTER_ID = '<идентификатор кластера>'JOB_NAME = 'clickhouse-spark-job'mlp = MLPlatform(REFRESH_TOKEN)logs = mlp.spark_job_logs(CLUSTER_ID, JOB_NAME)print(logs)

   Тапсырма сәтті орындалған кездегі шығару бөлігінің мысалы

2. (Опционалды) Кластердегі оқиғалар туралы ақпарат алыңыз. Ол, мысалы, ақауларды іздеу кезінде кластер мен тапсырмалардың ағымдағы күйін білуге мүмкіндік береді.

    
   from mlplatform_client import MLPlatformREFRESH_TOKEN = '<значение токена доступа>'CLUSTER_ID = '<идентификатор кластера>'mlp = MLPlatform(REFRESH_TOKEN)events = mlp.spark_events(CLUSTER_ID)print(events)

Егер жасалған ресурстар енді қажет болмаса, оларды жойыңыз:

1. Cloud Spark кластерін жойыңыз.
2. Осы кластерге арналған Docker тізілімін жойыңыз.
3. Осы кластер пайдаланған бакеттен объектілерді және бакеттің өзін жойыңыз.
4. Қолжетімділік токенін жойыңыз.
5. ClickHouse инстансын жойыңыз.