Spark Connect арқылы Spark кластеріне қосылу

Spark Connect Spark кластеріне қашықтан қосылып, онымен IDE арқылы жұмыс істеуге мүмкіндік береді.

1. VK Cloud жеке кабинетіне өтіңіз.

2. Жобаны таңдаңыз.

3. Егер әлі жасалмаған болса, Spark кластерін жасаңыз. Виртуалды машина түрін таңдаңыз, онда кемінде 6 CPU және 12 ГБ RAM болуы керек.

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

5. Қажетті деректер жиынын кластерге байланыстырылған VK Object Storage бакетіне орналастырыңыз:

   1. Объектілік қойма → Бакеттер бөліміне өтіңіз.
   2. Кластерге байланыстырылған бакетті таңдаңыз. Оның атауы <ИМЯ_КЛАСТЕРА>-<ДОМЕННОЕ_ИМЯ>-bucket маскасына сәйкес келеді.
   3. datasets директориясына өтіп, оған деректері бар файлды қосыңыз.
   4. Кейін пайдалану үшін бакет атауын сақтап қойыңыз.

6. ML Platform кітапханасы үшін әкімші құқықтары бар қолжетімділік токенін жасаңыз.

   Кейін пайдалану үшін токенді сақтап қойыңыз.

Spark Connect-пен жұмыс істеу үшін PySpark және пакеттің барлық тәуелділіктерін келесі тәсілдердің бірімен орнатыңыз:

 
pip3 install pysparkpip3 install pandaspip3 install pyarrowpip3 install grpciopip3 install protobufpip3 install grpcio-status

 
%pip install pyspark%pip install pandas%pip install pyarrow%pip install grpcio%pip install protobuf%pip install grpcio-status

Кластерге қосылып, бакетке жүктелген деректерді көрсету үшін келесі әрекеттерді орындаңыз:

1. Келесі мазмұндағы Python скриптін немесе Jupyter ноутбук файлын жасаңыз:

    
   from pyspark.sql import SparkSessionADMIN_REFRESH_TOKEN = "<ТОКЕН_ДОСТУПА>"spark_connect_url = f"sc://<ДОМЕННОЕ_ИМЯ_КЛАСТЕРА>:15002/;spark-token={ADMIN_REFRESH_TOKEN}"spark = SparkSession.builder.remote(spark_connect_url).appName("<ИМЯ_ПРИЛОЖЕНИЯ>").getOrCreate()df = spark.read.csv("s3a://<ИМЯ_БАКЕТА>/datasets/<ИМЯ_НАБОРА_ДАННЫХ>.csv", header=True, inferSchema=True)df.show()spark.stop()

   Мұнда:

   • <ТОКЕН_ДОСТУПА> — бұрын жасалған ML Platform кітапханасына қолжетімділік токені;
   • <ДОМЕННОЕ_ИМЯ_КЛАСТЕРА> — кластердің DNS атауы, мысалы: k8s-3d3b7fddd30040.ml.bizmrg.com;
   • <ИМЯ_ПРИЛОЖЕНИЯ> — кластер интерфейсінде көрсетілетін қолданба атауы;
   • <ИМЯ_БАКЕТА> — Spark кластерімен байланыстырылған VK Object Storage бакетінің атауы;
   • <ИМЯ_НАБОРА_ДАННЫХ> — бұрын VK Object Storage бакетіне жүктелген деректер жиынының атауы.

2. Скриптті IDE-ңызда ашып, оның орындалуын іске қосыңыз.

Нәтижесінде бұрын VK Object Storage бакетіне жүктелген жиын деректері көрсетіледі.

Cloud Spark кластерімен қашықтан жұмыс істеудің басқа мысалдарын сілтеме бойынша жүктеп ала аласыз.

1. Келесі мазмұндағы Python скриптін немесе Jupyter ноутбук файлын жасаңыз:

    
   from pyspark.sql import SparkSessionfrom datetime import datetime, datefrom pyspark.sql import RowADMIN_REFRESH_TOKEN = "<ТОКЕН_ДОСТУПА>"spark_connect_url = f"sc://<ДОМЕННОЕ_ИМЯ_КЛАСТЕРА>:15002/;spark-token={ADMIN_REFRESH_TOKEN}"spark = SparkSession.builder.remote(spark_connect_url).appName("<ИМЯ_ПРИЛОЖЕНИЯ>").getOrCreate()df = spark.createDataFrame([Row(a=1, b=2., c='string1', d=date(2000, 1, 1), e=datetime(2000, 1, 1, 12, 0)),Row(a=2, b=3., c='string2', d=date(2000, 2, 1), e=datetime(2000, 1, 2, 12, 0)),Row(a=4, b=5., c='string3', d=date(2000, 3, 1), e=datetime(2000, 1, 3, 12, 0))])df.show()spark.stop()

   Мұнда:

   • <ТОКЕН_ДОСТУПА> — бұрын жасалған ML Platform кітапханасына қолжетімділік токені;
   • <ДОМЕННОЕ_ИМЯ_КЛАСТЕРА> — кластердің DNS атауы, мысалы: k8s-3d3b7fddd30040.ml.bizmrg.com;
   • <ИМЯ_ПРИЛОЖЕНИЯ> — кластер интерфейсінде көрсетілетін қолданба атауы.

2. Скриптті IDE-ңызда ашып, оның орындалуын іске қосыңыз.

Нәтижесінде жасалған датафрейм қайтарылады:

 
+---+---+-------+----------+-------------------+|  a|  b|      c|         d|                  e|+---+---+-------+----------+-------------------+|  1|2.0|string1|2000-01-01|2000-01-01 07:00:00||  2|3.0|string2|2000-02-01|2000-01-02 07:00:00||  4|5.0|string3|2000-03-01|2000-01-03 07:00:00|+---+---+-------+----------+-------------------+

1. Келесі мазмұндағы Python скриптін немесе Jupyter ноутбук файлын жасаңыз:

    
   from pyspark.sql import SparkSessionADMIN_REFRESH_TOKEN = "<ТОКЕН_ДОСТУПА>"spark_connect_url = f"sc://<ДОМЕННОЕ_ИМЯ_КЛАСТЕРА>:15002/;spark-token={ADMIN_REFRESH_TOKEN}"spark = SparkSession.builder.remote(spark_connect_url).appName("<ИМЯ_ПРИЛОЖЕНИЯ>")     .config("spark.sql.legacy.setCommandRejectsSparkCoreConfs", "false")     .config("spark.jars.packages", "<НАЗВАНИЕ_ПАКЕТА_ИЗ_MAVEN>")     .getOrCreate()spark.stop()

   Мұнда:

   • <ТОКЕН_ДОСТУПА> — бұрын жасалған ML Platform кітапханасына қолжетімділік токені;
   • <ДОМЕННОЕ_ИМЯ_КЛАСТЕРА> — кластердің DNS атауы, мысалы: k8s-3d3b7fddd30040.ml.bizmrg.com;
   • <ИМЯ_ПРИЛОЖЕНИЯ> — кластер интерфейсінде көрсетілетін қолданба атауы;
   • <НАЗВАНИЕ_ПАКЕТА_ИЗ_MAVEN> — Maven репозиторийіндегі кітапхана атауы (jar-файл), мысалы: org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1.

2. Скриптті IDE-ңызда ашып, оның орындалуын іске қосыңыз.

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

Нәтижесінде қолданбаның жұмысына қажетті Maven пакеттері жүктеледі.