{"id":335296,"date":"2022-07-04T15:00:17","date_gmt":"2022-07-04T15:00:17","guid":{"rendered":"http:\/\/savepearlharbor.com\/?p=335296"},"modified":"-0001-11-30T00:00:00","modified_gmt":"-0001-11-29T21:00:00","slug":"","status":"publish","type":"post","link":"https:\/\/savepearlharbor.com\/?p=335296","title":{"rendered":"<span>\u041a\u0430\u043a \u0431\u044b\u0441\u0442\u0440\u043e \u0440\u0430\u0437\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438 \u0441 \u043f\u043e\u043c\u043e\u0449\u044c\u044e PySpark Structured Streaming \u0438 Kafka<\/span>"},"content":{"rendered":"<div><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w780q1\/getpro\/habr\/upload_files\/614\/3d3\/5b3\/6143d35b3d1e22a9d3891b1f44f0adf8.jpg\" width=\"780\" height=\"441\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/614\/3d3\/5b3\/6143d35b3d1e22a9d3891b1f44f0adf8.jpg\" data-blurred=\"true\"\/><figcaption><\/figcaption><\/figure>\n<p>\u0414\u0430\u043d\u043d\u0430\u044f \u0441\u0442\u0430\u0442\u044c\u044f \u043e\u0431\u043e\u0431\u0449\u0430\u0435\u0442 \u0431\u0430\u0437\u043e\u0432\u044b\u0435 \u0448\u0430\u0433\u0438 \u043f\u043e \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0435 \u0438 \u043d\u0430\u0447\u0430\u043b\u0443 \u0440\u0430\u0431\u043e\u0442\u044b \u0441 PySpark Structured Streaming \u043f\u0440\u0438 \u0443\u0447\u0430\u0441\u0442\u0438\u0438 \u0431\u0440\u043e\u043a\u0435\u0440\u0430 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 Kafka. \u041f\u0440\u0435\u0434\u043f\u043e\u043b\u0430\u0433\u0430\u0435\u0442\u0441\u044f, \u0447\u0442\u043e \u0447\u0438\u0442\u0430\u0442\u0435\u043b\u044c \u0443\u0436\u0435 \u0437\u043d\u0430\u043a\u043e\u043c \u0441 \u044f\u0437\u044b\u043a\u043e\u043c \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f Python \u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u043e\u043c Kafka.<\/p>\n<p>\u041f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 PySpark Structured Streaming \u043c\u043e\u0436\u043d\u043e \u0431\u044b\u0441\u0442\u0440\u043e \u0440\u0430\u0437\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u043c\u0430\u0441\u0448\u0442\u0430\u0431\u0438\u0440\u0443\u0435\u043c\u044b\u0435 \u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438. \u0422\u0430\u043a\u043e\u0439 \u043f\u043e\u0434\u0445\u043e\u0434 \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0432 \u043a\u043e\u0440\u043e\u0442\u043a\u0438\u0435 \u0441\u0440\u043e\u043a\u0438 \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u0432\u044b\u0433\u043e\u0434\u043d\u043e\u0435 \u043f\u0440\u0435\u0434\u043b\u043e\u0436\u0435\u043d\u0438\u0435 \u043a\u043b\u0438\u0435\u043d\u0442\u0443, \u0432\u043e\u0432\u0440\u0435\u043c\u044f \u0437\u0430\u043c\u0435\u0442\u0438\u0442\u044c \u0430\u043d\u043e\u043c\u0430\u043b\u0438\u044e \u0432 \u0441\u0438\u0441\u0442\u0435\u043c\u0435 \u0438\u043b\u0438 \u0436\u0435 \u043e\u0442\u043e\u0431\u0440\u0430\u0436\u0430\u0442\u044c \u0430\u043a\u0442\u0443\u0430\u043b\u044c\u043d\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435. \u041c\u0430\u0441\u0448\u0442\u0430\u0431\u0438\u0440\u0443\u0435\u043c\u043e\u0441\u0442\u044c \u043e\u0431\u0435\u0441\u043f\u0435\u0447\u0438\u0432\u0430\u0435\u0442\u0441\u044f \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u043e\u043c Spark. \u041c\u043e\u0434\u0443\u043b\u044c Structured Streaming \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0440\u0430\u0437\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u044b \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438, \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u0441\u0438\u043d\u0442\u0430\u043a\u0441\u0438\u0441 SQL.<\/p>\n<p>\u0417\u0434\u0435\u0441\u044c \u0431\u0443\u0434\u0443\u0442 \u0440\u0430\u0441\u0441\u043c\u043e\u0442\u0440\u0435\u043d\u044b \u0442\u043e\u043b\u044c\u043a\u043e \u0441\u0430\u043c\u044b\u0435 \u043f\u0440\u043e\u0441\u0442\u044b\u0435 \u043f\u0440\u0438\u043c\u0435\u0440\u044b \u0440\u0430\u0431\u043e\u0442\u044b \u0441 Kafka \u043f\u043e\u0441\u0440\u0435\u0434\u0441\u0442\u0432\u043e\u043c PySpark: \u0447\u0442\u0435\u043d\u0438\u0435\/\u0437\u0430\u043f\u0438\u0441\u044c \u0432 Kafka, \u0430 \u0442\u0430\u043a\u0436\u0435 \u0440\u0430\u0437\u0431\u043e\u0440 \u0438 \u0441\u043e\u0445\u0440\u0430\u043d\u0435\u043d\u0438\u0435 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0445 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 JSON \u0438 AVRO. \u0414\u043b\u044f \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e\u0433\u043e \u043e\u0437\u043d\u0430\u043a\u043e\u043c\u043b\u0435\u043d\u0438\u044f \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0441\u0442\u0430\u0442\u044c\u0438 \u043d\u0430 \u043e\u0444\u0438\u0446\u0438\u0430\u043b\u044c\u043d\u043e\u043c \u0441\u0430\u0439\u0442\u0435:<\/p>\n<ul>\n<li>\n<p><a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/structured-streaming-programming-guide.html\">Spark Structured Streaming<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/structured-streaming-kafka-integration.html\">Spark Structured Streaming Kafka Integration<\/a><\/p>\n<\/li>\n<\/ul>\n<h2>\u0422\u0440\u0435\u0431\u043e\u0432\u0430\u043d\u0438\u044f \u043a \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u044e<\/h2>\n<p>\u0412\u0441\u0435 \u043f\u0440\u0438\u043c\u0435\u0440\u044b \u0431\u044b\u043b\u0438 \u043e\u0442\u043b\u0430\u0436\u0435\u043d\u044b \u043d\u0430 \u043f\u043e\u0434\u0441\u0438\u0441\u0442\u0435\u043c\u0435 Ubuntu 20.04, Python 3.10.1, Spark 3.2.1, Kafka 3, OpenJDK 11.0.11 (\u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u0430 \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b Spark).<\/p>\n<h2>1.1) \u041f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u043a\u0430 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u0441\u0440\u0435\u0434\u044b<\/h2>\n<p>\u0421\u043e\u0437\u0434\u0430\u0451\u043c \u043f\u0443\u0441\u0442\u0443\u044e \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u044e \u0441 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435\u043c \u043f\u0440\u043e\u0435\u043a\u0442\u0430, \u0433\u0434\u0435 \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0438\u0440\u0443\u0435\u043c \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u0443\u044e \u0441\u0440\u0435\u0434\u0443 (\u043a\u043e\u043c\u0430\u043d\u0434\u0430 \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435):<\/p>\n<pre><code class=\"bash\">python -m venv .venv<\/code><\/pre>\n<p>\u0410\u043a\u0442\u0438\u0432\u0438\u0440\u0443\u0435\u043c \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435 \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0435\u0439 \u043a\u043e\u043c\u0430\u043d\u0434\u044b (\u0438\u0437 \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u0438, \u0433\u0434\u0435 \u0431\u044b\u043b\u0430 \u0441\u043e\u0437\u0434\u0430\u043d\u0430 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u0430\u044f \u0441\u0440\u0435\u0434\u0430):<\/p>\n<pre><code class=\"bash\">source .venv\/bin\/activate<\/code><\/pre>\n<p>\u0423\u0441\u0442\u0430\u043d\u0430\u0432\u043b\u0438\u0432\u0430\u0435\u043c PySpark:<\/p>\n<pre><code class=\"bash\">pip install wheel pyspark<\/code><\/pre>\n<p>\u0427\u0442\u043e\u0431\u044b \u043f\u0440\u043e\u0432\u0435\u0440\u0438\u0442\u044c, \u0447\u0442\u043e \u0432\u0441\u0451 \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u043e\u0441\u044c \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u043e, \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0438\u043d\u0442\u0435\u0440\u043f\u0440\u0435\u0442\u0430\u0442\u043e\u0440 PySpark:<\/p>\n<pre><code class=\"bash\">pyspark<\/code><\/pre>\n<p>\u0414\u043e\u043b\u0436\u0435\u043d \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c\u0441\u044f \u0438\u043d\u0442\u0435\u0440\u043f\u0440\u0435\u0442\u0430\u0442\u043e\u0440 Python, \u0433\u0434\u0435 \u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d Spark.<\/p>\n<figure class=\"bordered\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/61f\/058\/c24\/61f058c248b6ba83df69c7657be5d296.png\" alt=\"\u041f\u0440\u0438\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 Spark\" title=\"\u041f\u0440\u0438\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 Spark\" width=\"317\" height=\"115\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/61f\/058\/c24\/61f058c248b6ba83df69c7657be5d296.png\"\/><figcaption>\u041f\u0440\u0438\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 Spark<\/figcaption><\/figure>\n<h2>1.2) \u0417\u0430\u043f\u0443\u0441\u043a \u0441\u0435\u0440\u0432\u0438\u0441\u0430 Kafka<\/h2>\n<p>\u0414\u043b\u044f \u043f\u0440\u0438\u043c\u0435\u0440\u0430 Kafka \u0431\u0443\u0434\u0435\u0442 \u0437\u0430\u043f\u0443\u0449\u0435\u043d\u0430 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e. \u042d\u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0441\u0434\u0435\u043b\u0430\u0442\u044c, \u0432\u044b\u043f\u043e\u043b\u043d\u0438\u0432 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f \u0438\u0437 <a href=\"https:\/\/kafka.apache.org\/quickstart\">\u043e\u0444\u0438\u0446\u0438\u0430\u043b\u044c\u043d\u043e\u0439 \u0438\u043d\u0441\u0442\u0440\u0443\u043a\u0446\u0438\u0438<\/a>. \u0415\u0441\u043b\u0438 \u0443\u043c\u0435\u0435\u0442\u0435 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f Docker, \u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0435\u0451 \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u0434\u0430\u043d\u043d\u043e\u0433\u043e <a href=\"https:\/\/github.com\/bitnami\/bitnami-docker-kafka\/blob\/master\/docker-compose.yml\">docker-compose<\/a> \u0438\u043b\u0438 \u0436\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043b\u044e\u0431\u043e\u0439 \u0443\u0434\u043e\u0431\u043d\u044b\u0439 \u0434\u043b\u044f \u0432\u0430\u0441 \u043e\u0431\u0440\u0430\u0437.<\/p>\n<h2>2) \u0427\u0442\u0435\u043d\u0438\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka<\/h2>\n<p>\u0421\u043e\u0437\u0434\u0430\u0451\u043c \u043b\u044e\u0431\u043e\u0439 \u0444\u0430\u0439\u043b <code>*.py<\/code>. \u0414\u043b\u044f \u043f\u0440\u0438\u043c\u0435\u0440\u0430 \u044d\u0442\u043e \u0431\u0443\u0434\u0435\u0442 <code>example1.py<\/code>.<\/p>\n<p>\u0414\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0441\u043e Spark \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043e\u0431\u044a\u0435\u043a\u0442 <code>SparkSession<\/code>:<\/p>\n<pre><code class=\"python\">from pyspark.sql import SparkSession  spark = (SparkSession \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .builder \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .appName('quickstart-streaming-kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .getOrCreate()) spark.sparkContext.setLogLevel('WARN')<\/code><\/pre>\n<p>\u041f\u0440\u0438 \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 <code>SparkSession<\/code> \u0432\u044b \u043c\u043e\u0436\u0435\u0442\u0435 \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0440\u0430\u0437\u043b\u0438\u0447\u043d\u044b\u0435 \u0438\u043c\u0435\u043d\u0430 \u0432 \u043c\u0435\u0442\u043e\u0434\u0435 <code>appName()<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u00a0\u043f\u043e\u0437\u0432\u043e\u043b\u044f\u044e\u0442 \u043f\u043e\u0437\u0436\u0435 \u043e\u043f\u043e\u0437\u043d\u0430\u0432\u0430\u0442\u044c \u0441\u0432\u043e\u0438 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f (\u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u043f\u0440\u0438 \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433\u0435 \u0432\u0441\u0435\u0445 \u0437\u0430\u043f\u0443\u0449\u0435\u043d\u043d\u044b\u0445 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 Spark \u043d\u0430 \u043a\u043b\u0430\u0441\u0442\u0435\u0440\u0435).<\/p>\n<p>\u041f\u043e\u0441\u043b\u0435\u0434\u043d\u044f\u044f \u0441\u0442\u0440\u043e\u043a\u0430 \u043e\u043f\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u0430, \u043e\u043d\u0430 \u0432\u043b\u0438\u044f\u0435\u0442 \u043d\u0430 \u0432\u044b\u0432\u043e\u0434 \u043b\u043e\u0433\u043e\u0432 \u0440\u0430\u0431\u043e\u0442\u044b Spark. \u0423\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u0432 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435 <strong>&#171;WARN&#187;<\/strong>, \u043d\u0435 \u0431\u0443\u0434\u0443\u0442 \u0432\u044b\u0432\u043e\u0434\u0438\u0442\u044c\u0441\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u043e\u043d\u043d\u044b\u0435 \u043b\u043e\u0433\u0438.<\/p>\n<p>\u0423\u043a\u0430\u0436\u0435\u043c \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a \u0434\u0430\u043d\u043d\u044b\u0445 \u2013 \u0442\u043e\u043f\u0438\u043a Kafka:<\/p>\n<pre><code class=\"python\">source = (spark \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .readStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('subscribe', 'input00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .load())<\/code><\/pre>\n<p>\u0418\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043e\u0431\u044a\u0435\u043a\u0442 <code>SparkSession<\/code>, \u043c\u043e\u0436\u043d\u043e \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a \u0434\u0430\u043d\u043d\u044b\u0445, \u0431\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f \u0447\u0435\u043c\u0443 \u043f\u043e\u043b\u0443\u0447\u0438\u043c \u043e\u0431\u044a\u0435\u043a\u0442 <code>DataFrame<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0438 \u0431\u0443\u0434\u0435\u0442 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0442\u044c \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435. \u0418\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0438 \u0432 Spark \u0434\u0435\u043b\u044f\u0442\u0441\u044f \u043d\u0430 \u0434\u0432\u0430 \u0442\u0438\u043f\u0430:<\/p>\n<ul>\n<li>\n<p><strong>batch<\/strong> \u2013 \u0440\u0430\u0437\u043e\u0432\u0430\u044f \u0440\u0430\u0431\u043e\u0442\u0430, \u043a\u043e\u0433\u0434\u0430 \u0432\u0430\u043c \u0437\u0430\u0440\u0430\u043d\u0435\u0435 \u0438\u0437\u0432\u0435\u0441\u0442\u0435\u043d \u043e\u0431\u044a\u0451\u043c \u0434\u0430\u043d\u043d\u044b\u0445 (\u0434\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0441\u0432\u043e\u0439\u0441\u0442\u0432\u043e <code>read<\/code> \u0443 <code>SparkSession<\/code>);<\/p>\n<\/li>\n<li>\n<p><strong>stream<\/strong> \u2013 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u0430\u044f \u0440\u0430\u0431\u043e\u0442\u0430, \u043a\u043e\u0433\u0434\u0430 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435 \u0437\u0430\u043f\u0443\u0449\u0435\u043d\u043e \u0438 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0435\u0442 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0435 (\u0434\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0441\u0432\u043e\u0439\u0441\u0442\u0432\u043e <code>readStream<\/code>) \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438. \u0412 \u0434\u0430\u043d\u043d\u043e\u043c \u0440\u0435\u0436\u0438\u043c\u0435 Spark \u0441 \u043d\u0435\u043a\u043e\u0442\u043e\u0440\u043e\u0439 \u043f\u0435\u0440\u0438\u043e\u0434\u0438\u0447\u043d\u043e\u0441\u0442\u044c\u044e \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u0442 batch \u0438\u0437 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0445 \u0434\u0430\u043d\u043d\u044b\u0445.<\/p>\n<\/li>\n<\/ul>\n<p>\u0427\u0442\u043e\u0431\u044b \u0441\u0447\u0438\u0442\u044b\u0432\u0430\u0442\u044c \u0438\u0437 Kafka, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 <code>format()<\/code> \u043f\u0435\u0440\u0435\u0434\u0430\u0442\u044c \u0441\u0442\u0440\u043e\u043a\u0443 &#171;<strong>kafka<\/strong>&#171;. \u0421\u043f\u0438\u0441\u043e\u043a \u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u044b\u0445 \u043e\u043f\u0446\u0438\u0439 \u043f\u0440\u0438 \u0440\u0430\u0431\u043e\u0442\u0435 \u0441 \u0434\u0430\u043d\u043d\u044b\u043c \u0444\u043e\u0440\u043c\u0430\u0442\u043e\u043c:<\/p>\n<ul>\n<li>\n<p><strong>kafka.bootstrap.servers<\/strong> \u2013 \u0430\u0434\u0440\u0435\u0441 \u0441\u0435\u0440\u0432\u0435\u0440\u0430 \u0431\u0440\u043e\u043a\u0435\u0440\u0430 (\u043c\u043e\u0436\u043d\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u2013 \u0447\u0435\u0440\u0435\u0437 \u0437\u0430\u043f\u044f\u0442\u0443\u044e \u0432\u043d\u0443\u0442\u0440\u0438 \u0441\u0442\u0440\u043e\u043a\u0438);<\/p>\n<\/li>\n<li>\n<p><strong>subscribe<\/strong> \u2013 \u0442\u043e\u043f\u0438\u043a, \u043e\u0442\u043a\u0443\u0434\u0430 \u0431\u0443\u0434\u0443\u0442 \u0441\u0447\u0438\u0442\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u043d\u043e\u0432\u044b\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f (\u043c\u043e\u0436\u043d\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u2013 \u0447\u0435\u0440\u0435\u0437 \u0437\u0430\u043f\u044f\u0442\u0443\u044e \u0432\u043d\u0443\u0442\u0440\u0438 \u0441\u0442\u0440\u043e\u043a\u0438).<\/p>\n<\/li>\n<\/ul>\n<p>\u041f\u043e\u0441\u043c\u043e\u0442\u0440\u0438\u043c \u0441\u0445\u0435\u043c\u0443 \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c\u044b\u0445 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439, \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043c\u0435\u0442\u043e\u0434 <code>printSchema()<\/code> \u0443 \u043f\u043e\u043b\u0443\u0447\u0438\u0432\u0448\u0435\u0433\u043e\u0441\u044f <code>DataFrame<\/code>:<\/p>\n<pre><code class=\"python\">source.printSchema()<\/code><\/pre>\n<p>\u0412\u044b\u0432\u043e\u0434 \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435:<\/p>\n<pre><code>root \u00a0|-- key: binary (nullable = true) \u00a0|-- value: binary (nullable = true) \u00a0|-- topic: string (nullable = true) \u00a0|-- partition: integer (nullable = true) \u00a0|-- offset: long (nullable = true) \u00a0|-- timestamp: timestamp (nullable = true) \u00a0|-- timestampType: integer (nullable = true)<\/code><\/pre>\n<p>\u041a\u0440\u0430\u0442\u043a\u0430\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043f\u043e \u043a\u0430\u0436\u0434\u043e\u043c\u0443 \u043f\u043e\u043b\u044e:<\/p>\n<ol>\n<li>\n<p><strong>key<\/strong> \u2013 \u043a\u043b\u044e\u0447, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0432 Kafka \u0434\u043b\u044f \u043e\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u0438\u044f \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0438, \u0432 \u043a\u043e\u0442\u043e\u0440\u0443\u044e \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u043b\u043e \u043f\u043e\u043f\u0430\u0441\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435;<\/p>\n<\/li>\n<li>\n<p><strong>value<\/strong> \u2013 \u043d\u0435\u043f\u043e\u0441\u0440\u0435\u0434\u0441\u0442\u0432\u0435\u043d\u043d\u043e \u0441\u0430\u043c\u0438 \u0434\u0430\u043d\u043d\u044b\u0435, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c;<\/p>\n<\/li>\n<li>\n<p><strong>topic<\/strong> \u2013 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u0442\u043e\u043f\u0438\u043a\u0430, \u043e\u0442\u043a\u0443\u0434\u0430 \u0431\u044b\u043b\u043e \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 (\u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0438 \u0432 Spark \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u044e\u0442 \u043f\u043e\u0434\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u0441\u0440\u0430\u0437\u0443 \u043d\u0430 \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0442\u043e\u043f\u0438\u043a\u043e\u0432, \u0438\u0445 \u043c\u043e\u0436\u043d\u043e \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u0446\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043f\u043e \u0434\u0430\u043d\u043d\u043e\u043c\u0443 \u043f\u043e\u043b\u044e);<\/p>\n<\/li>\n<li>\n<p><strong>partition<\/strong> \u0438 <strong>offset<\/strong> \u2013 \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u044f \u0438 \u0441\u043c\u0435\u0449\u0435\u043d\u0438\u0435 \u0441\u043b\u0443\u0436\u0430\u0442 \u0443\u043a\u0430\u0437\u0430\u0442\u0435\u043b\u0435\u043c, \u0433\u0434\u0435 \u0438\u043c\u0435\u043d\u043d\u043e \u0432 \u0442\u043e\u043f\u0438\u043a\u0435 \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435;<\/p>\n<\/li>\n<li>\n<p><strong>timestamp<\/strong> \u0438 <strong>timestampType<\/strong> \u2013 \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u0432\u0440\u0435\u043c\u0435\u043d\u0438, \u043a\u043e\u0433\u0434\u0430 \u043f\u043e\u0441\u0442\u0443\u043f\u0438\u043b\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a.<\/p>\n<\/li>\n<\/ol>\n<p>\u041c\u043e\u0436\u043d\u043e \u0437\u0430\u043c\u0435\u0442\u0438\u0442\u044c, \u0447\u0442\u043e \u043f\u043e\u043b\u044f &#171;<strong>key<\/strong>&#187; \u0438 &#171;<strong>value<\/strong>&#187; \u0438\u043c\u0435\u044e\u0442 \u0434\u0432\u043e\u0438\u0447\u043d\u044b\u0439 \u0444\u043e\u0440\u043c\u0430\u0442. \u0418 \u0435\u0441\u043b\u0438 \u043f\u043e\u0441\u0442\u0443\u043f\u0438\u0442 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0441\u043e \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435\u043c &#171;<strong>Hello, World!<\/strong>&#171;, \u0442\u043e \u043e\u043d\u043e \u0431\u0443\u0434\u0435\u0442 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u043e \u043c\u0430\u0441\u0441\u0438\u0432\u043e\u043c \u0431\u0430\u0439\u0442\u043e\u0432:<\/p>\n<pre><code>------------------------------------------- Batch: 1 ------------------------------------------- +----+--------------------+-------+---------+------+--------------------+-------------+ | key|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 value|\u00a0 topic|partition|offset|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 timestamp|timestampType| +----+--------------------+-------+---------+------+--------------------+-------------+ |[30]|[48 65 6C 6C 6F 2...|input00|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 0|\u00a0\u00a0\u00a0\u00a0 3|2022-03-11 08:28:...|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 0| +----+--------------------+-------+---------+------+--------------------+-------------+<\/code><\/pre>\n<p>\u0412 Spark \u043c\u043e\u0436\u043d\u043e \u0432\u044b\u043f\u043e\u043b\u043d\u044f\u0442\u044c \u0432\u044b\u0431\u043e\u0440\u043a\u0443 \u0441 \u043f\u0440\u0438\u043c\u0435\u043d\u0435\u043d\u0438\u0435\u043c \u0440\u0430\u0437\u043b\u0438\u0447\u043d\u044b\u0445 \u0432\u044b\u0440\u0430\u0436\u0435\u043d\u0438\u0439, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u043f\u0440\u0438\u0432\u0435\u0434\u0435\u043d\u0438\u0435 \u0442\u0438\u043f\u043e\u0432. \u0418\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043c\u0435\u0442\u043e\u0434 <code>selectExpr()<\/code>, \u0433\u0434\u0435 \u043c\u043e\u0436\u043d\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0432\u044b\u0432\u043e\u0434\u0438\u043c\u044b\u0435 \u043f\u043e\u043b\u044f \u0432 <code>DataFrame<\/code>, \u0432\u044b\u0432\u0435\u0434\u0435\u043c \u0438 \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u0443\u0435\u043c \u043f\u043e\u043b\u0435 &#171;<strong>value<\/strong>&#187; \u043a \u0441\u0442\u0440\u043e\u043a\u0435, \u0430 \u0442\u0430\u043a\u0436\u0435 \u0432\u044b\u0432\u0435\u0434\u0435\u043c &#171;<strong>offset<\/strong>&#171;:<\/p>\n<pre><code class=\"python\">df = (source \u00a0\u00a0\u00a0\u00a0\u00a0 .selectExpr('CAST(value AS STRING)', 'offset'))<\/code><\/pre>\n<p>\u041d\u0430\u043f\u0438\u0448\u0435\u043c \u043f\u0440\u043e\u0441\u0442\u043e\u0439 \u0432\u044b\u0432\u043e\u0434 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0430 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c:<\/p>\n<pre><code class=\"python\">console = (df            .writeStream            .format('console'))<\/code><\/pre>\n<p>\u0423 \u043b\u044e\u0431\u043e\u0433\u043e <code>DataFrame<\/code> \u043c\u043e\u0436\u043d\u043e \u0432\u044b\u0437\u0432\u0430\u0442\u044c \u0441\u0432\u043e\u0439\u0441\u0442\u0432\u043e <code>writeStream<\/code>, \u043f\u043e\u0441\u043b\u0435 \u0447\u0435\u0433\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0444\u043e\u0440\u043c\u0430\u0442 \u0432\u044b\u0432\u043e\u0434\u0430. \u00a0\u0412 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u044d\u0442\u043e \u0432\u044b\u0432\u043e\u0434 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c (\u043f\u043e\u043b\u043d\u044b\u0439 \u0441\u043f\u0438\u0441\u043e\u043a \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0435\u043c\u044b\u0445 \u0444\u043e\u0440\u043c\u0430\u0442\u043e\u0432 \u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d \u0432 <a href=\"https:\/\/spark.apache.org\/docs\/latest\/structured-streaming-programming-guide.html#output-sinks\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>). \u0422\u0430\u043a\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c, \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u0441\u044f \u043e\u0431\u044a\u0435\u043a\u0442 <code>DataStreamWriter<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043e\u0441\u0442\u0430\u043b\u043e\u0441\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c:<\/p>\n<pre><code class=\"python\">console.start().awaitTermination()<\/code><\/pre>\n<p>\u041c\u0435\u0442\u043e\u0434 <code>start()<\/code> \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u0442 \u043f\u043e\u0442\u043e\u043a \u043d\u0430 \u0432\u044b\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u0435. \u041d\u043e \u0434\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0430 \u043d\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u0430\u0441\u044c, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043f\u043e\u0441\u0442\u0430\u0432\u0438\u0442\u044c \u0435\u0451 \u0432 \u043e\u0436\u0438\u0434\u0430\u043d\u0438\u0435. \u042d\u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u043c\u0435\u0442\u043e\u0434\u043e\u043c <code>awaitTermination()<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0431\u0443\u0434\u0435\u0442 \u043e\u0436\u0438\u0434\u0430\u0442\u044c, \u043f\u043e\u043a\u0430 \u043f\u043e\u0442\u043e\u043a \u043d\u0435 \u0431\u0443\u0434\u0435\u0442 \u0442\u0435\u0440\u043c\u0438\u043d\u0438\u0440\u043e\u0432\u0430\u043d.<\/p>\n<p>\u041f\u043e\u0441\u043c\u043e\u0442\u0440\u0438\u043c \u043d\u0430 \u0432\u044b\u0432\u043e\u0434, \u0437\u0430\u043f\u0438\u0441\u0430\u0432 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 &#171;<strong>Hello, World!<\/strong>&#187; \u0432 \u0442\u043e\u043f\u0438\u043a &#171;<strong>input00<\/strong>&#171;:<\/p>\n<pre><code>------------------------------------------- Batch: 1 ------------------------------------------- +-------------+------+ |\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 value|offset| +-------------+------+ |Hello, World!|\u00a0\u00a0\u00a0\u00a0 5| +-------------+------+<\/code><\/pre>\n<p>\u0413\u043e\u0442\u043e\u0432\u043e! \u0422\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u0447\u0438\u0442\u0430\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0438\u0437 Kafka.<\/p>\n<details class=\"spoiler\">\n<summary>\u041f\u043e\u043b\u043d\u044b\u0439 \u043a\u043e\u0434 \u043f\u0440\u0438\u043c\u0435\u0440\u0430<\/summary>\n<div class=\"spoiler__content\">\n<pre><code class=\"python\">def main(): \u00a0\u00a0\u00a0 from pyspark.sql import SparkSession  \u00a0\u00a0\u00a0 spark = (SparkSession \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .builder \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .appName('streaming-kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .getOrCreate()) \u00a0\u00a0\u00a0 spark.sparkContext.setLogLevel('WARN')  \u00a0\u00a0\u00a0 source = (spark \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .readStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('subscribe', 'input00,input01') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .load()) \u00a0\u00a0\u00a0 source.printSchema()  \u00a0\u00a0\u00a0 df = (source \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .selectExpr('CAST(value AS STRING)', 'offset'))  \u00a0\u00a0\u00a0 console = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('console') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \u00a0\u00a0\u00a0\u00a0\u00a0.queryName('console output')) \u00a0\u00a0\u00a0 console.start().awaitTermination()   if __name__ == '__main__': \u00a0\u00a0\u00a0 main()<\/code><\/pre>\n<\/p>\n<\/div>\n<\/details>\n<h2>3) \u0417\u0430\u043f\u0443\u0441\u043a \u0441\u043a\u0440\u0438\u043f\u0442\u0430<\/h2>\n<p>\u0414\u043b\u044f \u0437\u0430\u043f\u0443\u0441\u043a\u0430 Spark \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f <strong>spark-submit<\/strong>. \u0427\u0442\u043e\u0431\u044b \u0432\u0437\u0430\u0438\u043c\u043e\u0434\u0435\u0439\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c \u0441 Kafka, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0434\u043e\u0431\u0430\u0432\u0438\u0442\u044c \u0437\u0430\u0432\u0438\u0441\u0438\u043c\u043e\u0441\u0442\u044c &#171;<strong>spark-sql-kafka-0-10_2.12<\/strong>&#187; (\u0430\u043a\u0442\u0443\u0430\u043b\u044c\u043d\u0430\u044f \u0432\u0435\u0440\u0441\u0438\u044f \u0434\u043b\u044f Spark 3.2.1 \u043d\u0430 \u043c\u043e\u043c\u0435\u043d\u0442 \u043d\u0430\u043f\u0438\u0441\u0430\u043d\u0438\u044f \u0441\u0442\u0430\u0442\u044c\u0438) \u043a \u0430\u0440\u0433\u0443\u043c\u0435\u043d\u0442\u0443 &#171;<strong>&#8212;packages<\/strong>&#187; (\u0435\u0441\u043b\u0438 \u0435\u0441\u0442\u044c \u0434\u043e\u0441\u0442\u0443\u043f \u043a \u0441\u0435\u0442\u0438 \u0418\u043d\u0442\u0435\u0440\u043d\u0435\u0442 \u0438\u043b\u0438 \u043f\u0430\u043a\u0435\u0442 \u0443\u0436\u0435 \u0438\u043c\u0435\u0435\u0442\u0441\u044f \u0432 \u043a\u044d\u0448\u0435) \u0438\u043b\u0438 &#171;<strong>&#8212;jars<\/strong>&#187; (\u0435\u0441\u043b\u0438 \u0438\u043c\u0435\u0435\u0442\u0441\u044f \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e \u043d\u0430 \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u043c\u043e\u0439 \u043c\u0430\u0448\u0438\u043d\u0435, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u043f\u0443\u0442\u044c).<\/p>\n<p><strong>spark-submit<\/strong> \u043f\u0440\u0438\u043d\u0438\u043c\u0430\u0435\u0442 \u043c\u043d\u043e\u0436\u0435\u0441\u0442\u0432\u043e \u043f\u0430\u0440\u0430\u043c\u0435\u0442\u0440\u043e\u0432, \u043e \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0432 <a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/submitting-applications.html\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>. \u042d\u0442\u0430 \u0443\u0442\u0438\u043b\u0438\u0442\u0430 \u0434\u043e\u0441\u0442\u0443\u043f\u043d\u0430 \u0432 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u0441\u0440\u0435\u0434\u0435 Python, \u0433\u0434\u0435 \u0431\u044b\u043b \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d \u043f\u0430\u043a\u0435\u0442 PySpark, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 \u0435\u0451 \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u043e\u0439 \u043a\u043e\u043c\u0430\u043d\u0434\u043e\u0439 \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435 \u043f\u0440\u0438 \u0430\u043a\u0442\u0438\u0432\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0439 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u0441\u0440\u0435\u0434\u0435 (\u0434\u043e\u043b\u0436\u043d\u0430 \u0431\u044b\u0442\u044c \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d\u0430 Java):<\/p>\n<pre><code class=\"bash\">spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 example1.py<\/code><\/pre>\n<h2>4) \u041f\u0438\u0448\u0435\u043c \u0432 \u0442\u043e\u043f\u0438\u043a Kafka<\/h2>\n<p>\u0412\u044b\u0432\u043e\u0434 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka \u043d\u0430\u043f\u0438\u0441\u0430\u0442\u044c \u043d\u0435 \u043c\u043d\u043e\u0433\u0438\u043c \u0441\u043b\u043e\u0436\u043d\u0435\u0435, \u0447\u0435\u043c \u0432\u044b\u0432\u043e\u0434 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c.<\/p>\n<p>\u0414\u043e\u043f\u043e\u043b\u043d\u044f\u044f \u043f\u0440\u0435\u0434\u044b\u0434\u0443\u0449\u0438\u0439 \u043f\u0440\u0438\u043c\u0435\u0440, \u043d\u0430\u043f\u0438\u0448\u0435\u043c \u0435\u0449\u0451 \u043e\u0434\u0438\u043d \u0432\u044b\u0445\u043e\u0434 \u0434\u043b\u044f <code>df<\/code>:<\/p>\n<pre><code class=\"python\">query = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('kafka-output') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('topic', 'output00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('checkpointLocation', '.\/.local\/checkpoint'))<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0432\u044b\u0432\u043e\u0434\u0430 \u0432 Kafka \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432 \u043c\u0435\u0442\u043e\u0434 <code>format()<\/code> \u0437\u0430\u043f\u0438\u0441\u0430\u0442\u044c &#171;<strong>kafka<\/strong>&#187; \u0438 \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0432 \u043e\u043f\u0446\u0438\u044f\u0445:<\/p>\n<ul>\n<li>\n<p><strong>kafka.bootstrap.servers <\/strong>\u2013 \u0430\u0434\u0440\u0435\u0441 \u0431\u0440\u043e\u043a\u0435\u0440 \u0441\u0435\u0440\u0432\u0435\u0440\u0430 (\u043c\u043e\u0436\u043d\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u2013 \u00a0\u0447\u0435\u0440\u0435\u0437 \u0437\u0430\u043f\u044f\u0442\u0443\u044e \u0432\u043d\u0443\u0442\u0440\u0438 \u0441\u0442\u0440\u043e\u043a\u0438);<\/p>\n<\/li>\n<li>\n<p><strong>topic<\/strong> \u2013 \u00a0\u0432 \u043a\u0430\u043a\u043e\u0439 \u0442\u043e\u043f\u0438\u043a \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 (\u0442\u043e\u043b\u044c\u043a\u043e \u043e\u0434\u0438\u043d);<\/p>\n<\/li>\n<li>\n<p><strong>checkpointLocation<\/strong> \u2013 \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u044e, \u0433\u0434\u0435 \u0431\u0443\u0434\u0443\u0442 \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c\u0441\u044f\/\u0447\u0438\u0442\u0430\u0442\u044c\u0441\u044f \u0447\u0435\u043a\u043f\u043e\u0438\u043d\u0442\u044b. \u0427\u0435\u043a\u043f\u043e\u0438\u043d\u0442\u044b \u0441\u043e\u0445\u0440\u0430\u043d\u044f\u044e\u0442 \u0443\u043a\u0430\u0437\u0430\u0442\u0435\u043b\u044c \u043d\u0430 \u043c\u0435\u0441\u0442\u0435, \u0433\u0434\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u0430\u0441\u044c \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0430 \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0430, \u0447\u0442\u043e \u043c\u043e\u0436\u0435\u0442 \u043f\u0440\u0438\u0433\u043e\u0434\u0438\u0442\u044c\u0441\u044f \u043f\u0440\u0438 \u043f\u0435\u0440\u0435\u0437\u0430\u043f\u0443\u0441\u043a\u0435 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u044b (\u043f\u0440\u043e\u0434\u043e\u043b\u0436\u0438\u0442 \u0441 \u0442\u043e\u0433\u043e \u043c\u0435\u0441\u0442\u0430, \u0433\u0434\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u0441\u044f). \u041d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u0435\u0442\u0441\u044f \u043d\u0430 <code>DataStreamWriter<\/code>. \u0411\u043e\u043b\u0435\u0435 \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e \u043f\u0440\u043e \u0447\u0435\u043a\u043f\u043e\u0438\u043d\u0442\u044b \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0432 <a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>.<\/p>\n<\/li>\n<\/ul>\n<p>\u0415\u0449\u0451 \u0434\u043e\u0431\u0430\u0432\u043e\u0447\u043d\u043e \u0437\u0434\u0435\u0441\u044c \u043f\u043e\u044f\u0432\u0438\u043b\u0430\u0441\u044c \u0441\u0442\u0440\u043e\u043a\u0430 \u0441 \u043c\u0435\u0442\u043e\u0434\u043e\u043c <code>queryName()<\/code>. \u042d\u0442\u043e \u043d\u0435\u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e, \u043d\u043e \u0442\u0430\u043a \u043a\u0430\u043a \u0443 \u043d\u0430\u0441 \u043f\u043e\u044f\u0432\u0438\u043b\u0441\u044f \u0432\u0442\u043e\u0440\u043e\u0439 \u043f\u043e\u0442\u043e\u043a \u0432 \u043e\u0434\u043d\u043e\u0439 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0435, \u0441\u043b\u0435\u0434\u0443\u0435\u0442 \u0438\u0445 \u0432\u0441\u0435 \u043f\u0440\u043e\u0438\u043c\u0435\u043d\u043e\u0432\u0430\u0442\u044c, \u0447\u0442\u043e\u0431\u044b \u0440\u0430\u0437\u043b\u0438\u0447\u0430\u0442\u044c \u043f\u0440\u0438 \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433\u0435. \u0417\u0430\u043f\u0443\u0441\u0442\u0438\u043c \u043e\u0431\u0430 \u043f\u043e\u0442\u043e\u043a\u0430:<\/p>\n<pre><code class=\"python\">console.start() query.start().awaitTermination()<\/code><\/pre>\n<p>\u0412\u044b\u0432\u043e\u0434:<\/p>\n<pre><code>------------------------------------------- Batch: 1 ------------------------------------------- +-------------------+------+ |\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 value|offset| +-------------------+------+ |Hello, second sink!|\u00a0\u00a0\u00a0\u00a0 0| +-------------------+------+<\/code><\/pre>\n<p>\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka \u0442\u0430\u043a\u0436\u0435 \u0437\u0430\u043f\u0438\u0441\u0430\u043d:<\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/6f6\/463\/a50\/6f6463a50b4c82a618788b93520feb34.png\" alt=\"\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u043c \u0442\u043e\u043f\u0438\u043a\u0435 Kafka\" title=\"\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u043c \u0442\u043e\u043f\u0438\u043a\u0435 Kafka\" width=\"1363\" height=\"220\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/6f6\/463\/a50\/6f6463a50b4c82a618788b93520feb34.png\"\/><figcaption>\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u043c \u0442\u043e\u043f\u0438\u043a\u0435 Kafka<\/figcaption><\/figure>\n<p><strong>\u0412\u0410\u0416\u041d\u041e<\/strong>: \u0443 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u0433\u043e <code>DataFrame<\/code> \u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u0442\u044c \u043f\u043e\u043b\u0435 &#171;<strong>value<\/strong>&#171;, \u0438\u043d\u0430\u0447\u0435 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0430 \u0443\u043f\u0430\u0434\u0451\u0442 \u0441 \u043e\u0448\u0438\u0431\u043a\u043e\u0439. \u0418\u043c\u0435\u043d\u043d\u043e \u044d\u0442\u043e \u043f\u043e\u043b\u0435 \u0438 \u0431\u0443\u0434\u0435\u0442 \u0437\u0430\u043f\u0438\u0441\u0430\u043d\u043e \u0432 &#171;<strong>value<\/strong>&#187; \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0434\u043b\u044f Kafka. \u041f\u043e\u043b\u0435 &#171;<strong>key<\/strong>&#187; \u043c\u043e\u0436\u0435\u0442 \u043e\u0442\u0441\u0443\u0442\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c \u0432 <code>DataFrame<\/code>.<\/p>\n<p>\u0411\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f \u0442\u043e\u043c\u0443, \u0447\u0442\u043e \u0443 \u043a\u0430\u0436\u0434\u043e\u0433\u043e \u043f\u043e\u0442\u043e\u043a\u0430 \u0435\u0441\u0442\u044c \u0441\u0432\u043e\u0435 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435, \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u0438\u0445 \u0441 \u043b\u0451\u0433\u043a\u043e\u0441\u0442\u044c\u044e \u043e\u0442\u043b\u0438\u0447\u0430\u0442\u044c \u0432 <strong>Spark Web-UI<\/strong>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u0442\u0441\u044f \u043f\u0430\u0440\u0430\u043b\u043b\u0435\u043b\u044c\u043d\u043e \u0441 \u0440\u0430\u0437\u0432\u0451\u0440\u0442\u044b\u0432\u0430\u043d\u0438\u0435\u043c Spark \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f (\u043f\u043e \u0443\u043c\u043e\u043b\u0447\u0430\u043d\u0438\u044e \u0434\u043e\u0441\u0442\u0443\u043f\u043d\u043e \u043f\u043e <a href=\"https:\/\/localhost:4040\">localhost:4040<\/a>).<\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/105\/024\/9bd\/1050249bd7c49780e21dcb65b8ac2f4e.png\" alt=\"Spark Web-UI\" title=\"Spark Web-UI\" width=\"1366\" height=\"481\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/105\/024\/9bd\/1050249bd7c49780e21dcb65b8ac2f4e.png\"\/><figcaption>Spark Web-UI<\/figcaption><\/figure>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/d7a\/f19\/153\/d7af191530a88b39e02391cbd9701d61.png\" alt=\"\u041e\u043a\u043d\u043e \u0441\u0442\u0430\u0442\u0438\u0441\u0442\u0438\u043a\u0438 \u0432\u044b\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u044f \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 Spark\" title=\"\u041e\u043a\u043d\u043e \u0441\u0442\u0430\u0442\u0438\u0441\u0442\u0438\u043a\u0438 \u0432\u044b\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u044f \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 Spark\" width=\"1366\" height=\"688\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/d7a\/f19\/153\/d7af191530a88b39e02391cbd9701d61.png\"\/><figcaption>\u041e\u043a\u043d\u043e \u0441\u0442\u0430\u0442\u0438\u0441\u0442\u0438\u043a\u0438 \u0432\u044b\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u044f \u0437\u0430\u043f\u0440\u043e\u0441\u043e\u0432 Spark<\/figcaption><\/figure>\n<p>\u041e\u0442\u043b\u0438\u0447\u043d\u043e, \u0442\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c \u043d\u043e\u0432\u044b\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0432 \u0442\u043e\u043f\u0438\u043a Kafka!<\/p>\n<details class=\"spoiler\">\n<summary>\u041f\u043e\u043b\u043d\u044b\u0439 \u043a\u043e\u0434 \u043f\u0440\u0438\u043c\u0435\u0440\u0430<\/summary>\n<div class=\"spoiler__content\">\n<pre><code class=\"python\">def main():     from pyspark.sql import SparkSession      spark = (SparkSession              .builder              .appName('streaming-kafka')              .getOrCreate())     spark.sparkContext.setLogLevel('WARN')      source = (spark               .readStream               .format('kafka')               .option('kafka.bootstrap.servers', 'localhost:9092')               .option('subscribe', 'input00')               .load())      df = (source           .selectExpr('CAST(value AS STRING)', 'offset'))      console = (df                .writeStream                .format('console')                .queryName('console-output'))      query = (df              .writeStream              .format('kafka')              .queryName('kafka-output')              .option('kafka.bootstrap.servers', 'localhost:9092')              .option('topic', 'output00')              .option('checkpointLocation', '.\/.local\/checkpoint'))      console.start()     query.start().awaitTermination()   if __name__ == '__main__':     main()<\/code><\/pre>\n<\/p>\n<\/div>\n<\/details>\n<h2>5) \u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 JSON<\/h2>\n<p>\u041f\u0435\u0440\u0435\u0434\u0430\u0432\u0430\u044f \u0434\u0430\u043d\u043d\u044b\u0435, \u043c\u044b \u0432\u0441\u0435\u0433\u0434\u0430 \u0436\u0434\u0451\u043c \u0447\u0435\u0433\u043e-\u0442\u043e \u0431\u043e\u043b\u044c\u0448\u0435\u0433\u043e, \u0447\u0435\u043c \u043f\u0440\u043e\u0441\u0442\u043e \u043e\u0434\u043d\u0443 \u0441\u0442\u0440\u043e\u043a\u0443. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u043c\u043e\u0436\u043d\u043e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u0442\u0435\u043a\u0441\u0442\u043e\u0432\u044b\u0435 \u0444\u043e\u0440\u043c\u0430\u0442\u044b \u0434\u0430\u043d\u043d\u044b\u0445, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, JSON. Spark \u0438\u043c\u0435\u0435\u0442 \u0444\u0443\u043d\u043a\u0446\u0438\u0438, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043c\u043e\u0433\u0443\u0442 \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u043e\u0432\u0430\u0442\u044c \u0441\u0442\u0440\u043e\u043a\u0443 \u0432 <code>struct<\/code>, \u0442\u043e \u0435\u0441\u0442\u044c \u0432 \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443 \u0434\u0430\u043d\u043d\u044b\u0445.<\/p>\n<p>\u0411\u0443\u0434\u0435\u043c \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0443\u044e \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443 \u0434\u0430\u043d\u043d\u044b\u0445:<\/p>\n<pre><code>{     \"name\": String, \u00a0\u00a0\u00a0\u00a0\"age\": Int }<\/code><\/pre>\n<p>\u0417\u0430\u0432\u043e\u0434\u0438\u043c \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a <code>source<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0432\u044b\u0447\u0438\u0442\u044b\u0432\u0430\u0435\u0442 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka, \u043f\u0440\u0438 \u044d\u0442\u043e\u043c \u043f\u043e\u043b\u0435 &#171;<strong>value<\/strong>&#187; \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u043c \u043a \u0442\u0438\u043f\u0443 <code>String<\/code> (\u043a\u0430\u043a \u043e\u043f\u0438\u0441\u0430\u043d\u043e \u0432 \u0440\u0430\u0437\u0434\u0435\u043b\u0435 2).<\/p>\n<p>\u0412 Spark \u0435\u0441\u0442\u044c \u0444\u0443\u043d\u043a\u0446\u0438\u044f <code>from_json()<\/code>, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u043c\u043e\u0436\u0435\u0442 \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u0442\u044c \u043f\u043e\u043b\u0435 \u0438\u0437 \u0441\u0442\u0440\u043e\u043a\u0438 \u043a \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0435. \u041f\u0440\u0438 \u044d\u0442\u043e\u043c \u0435\u0439 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043f\u0435\u0440\u0435\u0434\u0430\u0442\u044c \u0441\u0445\u0435\u043c\u0443 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f, \u043a\u043e\u0442\u043e\u0440\u0430\u044f \u043c\u043e\u0436\u0435\u0442 \u0431\u044b\u0442\u044c \u0442\u0438\u043f\u0430 <code>DataType<\/code>, \u0430 \u0442\u0430\u043a\u0436\u0435 <code>String<\/code>. \u0412\u043e\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u043c\u0441\u044f \u043f\u0435\u0440\u0432\u044b\u043c \u0432\u0430\u0440\u0438\u0430\u043d\u0442\u043e\u043c:<\/p>\n<pre><code class=\"python\">from pyspark.sql import types as t  schema = t.StructType( \u00a0\u00a0\u00a0 [ \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 t.StructField('name', t.StringType(), True), \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 t.StructField('age', t.IntegerType(), True), \u00a0\u00a0\u00a0 ], )<\/code><\/pre>\n<p>\u0412\u0441\u0435 \u0442\u0438\u043f\u044b \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0439\u0442\u0438 \u0432 \u043f\u0430\u043a\u0435\u0442\u0435 <code>pyspark.sql.types<\/code>. \u041a\u043e\u043c\u0431\u0438\u043d\u0438\u0440\u0443\u044f \u0442\u0430\u043a\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c \u0442\u0438\u043f\u044b \u0434\u0430\u043d\u043d\u044b\u0445, \u043c\u043e\u0436\u043d\u043e \u043e\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u0443\u044e \u0441\u0445\u0435\u043c\u0443.<\/p>\n<p>\u0422\u0435\u043f\u0435\u0440\u044c \u043d\u0443\u0436\u043d\u043e \u043f\u0440\u0438\u043c\u0435\u043d\u0438\u0442\u044c \u0444\u0443\u043d\u043a\u0446\u0438\u044e <code>from_json()<\/code> \u043a \u043f\u043e\u043b\u044e, \u043a\u043e\u0442\u043e\u0440\u043e\u0435 \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u0442 \u0441\u0442\u0440\u043e\u043a\u0443 \u0441 JSON:<\/p>\n<pre><code class=\"python\">from pyspark.sql import functions as f  df = (df \u00a0\u00a0\u00a0\u00a0\u00a0 .select(f.from_json('value', schema).alias('data'))) print(\"df schema\") df.printSchema()<\/code><\/pre>\n<p>\u0412\u044b\u0432\u043e\u0434 \u0441\u0445\u0435\u043c\u044b:<\/p>\n<pre><code>df schema root \u00a0|-- data: struct (nullable = true) \u00a0|\u00a0\u00a0\u00a0 |-- name: string (nullable = true) \u00a0|\u00a0\u00a0\u00a0 |-- age: integer (nullable = true)<\/code><\/pre>\n<p>\u041f\u0435\u0440\u0432\u044b\u043c \u0430\u0440\u0433\u0443\u043c\u0435\u043d\u0442\u043e\u043c \u0444\u0443\u043d\u043a\u0446\u0438\u0438 <code>from_json()<\/code> \u0431\u044b\u043b\u043e \u043f\u0435\u0440\u0435\u0434\u0430\u043d\u043e \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u043f\u043e\u043b\u044f, \u0432\u0442\u043e\u0440\u044b\u043c \u0430\u0440\u0433\u0443\u043c\u0435\u043d\u0442\u043e\u043c \u2013 \u0441\u0445\u0435\u043c\u0430 \u0434\u0430\u043d\u043d\u044b\u0445. \u041c\u0435\u0442\u043e\u0434 <code>alias()<\/code> \u0438\u0437\u043c\u0435\u043d\u044f\u0435\u0442 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u043f\u043e\u043b\u0443\u0447\u0438\u0432\u0448\u0435\u0433\u043e\u0441\u044f \u043f\u043e\u043b\u044f (\u0430\u043d\u0430\u043b\u043e\u0433\u0438\u0447\u043d\u043e \u0441\u043b\u043e\u0432\u0443 <code>AS<\/code> \u0432 SQL).<\/p>\n<p>\u0414\u043e\u0431\u0430\u0432\u0438\u043c \u043d\u0435\u043c\u043d\u043e\u0433\u043e \u043b\u043e\u0433\u0438\u043a\u0438: \u0431\u0443\u0434\u0435\u043c \u0444\u0438\u043b\u044c\u0442\u0440\u043e\u0432\u0430\u0442\u044c \u043f\u043e \u043f\u043e\u043b\u044e &#171;<strong>age<\/strong>&#171;, \u0447\u0442\u043e \u0431\u043e\u043b\u044c\u0448\u0435 \u0438\u043b\u0438 \u0440\u0430\u0432\u043d\u043e 18 \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u043c\u0435\u0442\u043e\u0434\u0430 <code>where()<\/code>. \u0422\u0430\u043a \u043a\u0430\u043a \u043f\u0440\u0438 \u043f\u0430\u0440\u0441\u0438\u043d\u0433\u0435 \u043c\u044b \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443, \u0442\u043e \u043f\u0440\u0438 \u043e\u0431\u0440\u0430\u0449\u0435\u043d\u0438\u0438 \u043a \u043a\u043e\u043d\u043a\u0440\u0435\u0442\u043d\u043e\u043c\u0443 \u043f\u043e\u043b\u044e \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u044b \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043e\u0431\u0440\u0430\u0449\u0430\u0442\u044c\u0441\u044f \u0447\u0435\u0440\u0435\u0437 \u0442\u043e\u0447\u043a\u0443 (&#171;<strong>data.age<\/strong>&#171;):<\/p>\n<pre><code class=\"python\">df = df.where(df['data.age'] >= 18)<\/code><\/pre>\n<p>\u0421\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443 \u043c\u043e\u0436\u043d\u043e \u043b\u0435\u0433\u043a\u043e \u0440\u0430\u0437\u0432\u0435\u0440\u043d\u0443\u0442\u044c \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u043c\u0435\u0442\u043e\u0434\u0430 <code>select()<\/code>:<\/p>\n<pre><code class=\"python\">console = df.select('data.*') print(\"console schema\") console.printSchema()<\/code><\/pre>\n<p>\u0412\u044b\u0432\u043e\u0434 \u0441\u0445\u0435\u043c\u044b:<\/p>\n<pre><code>console schema root \u00a0|-- name: string (nullable = true) \u00a0|-- age: integer (nullable = true)<\/code><\/pre>\n<p>\u041d\u0435 \u0441\u0442\u043e\u0438\u0442 \u043f\u0435\u0440\u0435\u0436\u0438\u0432\u0430\u0442\u044c \u043f\u043e \u043f\u043e\u0432\u043e\u0434\u0443 \u0441\u0442\u0440\u043e\u043a \u0441 \u0432\u044b\u0432\u043e\u0434\u043e\u043c \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c, \u043a\u0430\u043a <code>print()<\/code> \u0438 <code>printSchema()<\/code>, \u0442\u0430\u043a \u043a\u0430\u043a \u0432 \u0441\u0430\u043c \u0437\u0430\u043f\u0440\u043e\u0441 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 Spark \u043e\u043d\u0438 \u043d\u0435 \u043f\u043e\u043f\u0430\u0434\u0443\u0442, \u0432\u044b\u043f\u043e\u043b\u043d\u044f\u0442\u0441\u044f\u00a0\u0442\u043e\u043b\u044c\u043a\u043e \u0432 \u043c\u043e\u043c\u0435\u043d\u0442 \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0437\u0430\u043f\u0440\u043e\u0441\u0430 (\u0442\u043e \u0435\u0441\u0442\u044c \u043f\u0440\u0438 \u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0438 \u043d\u043e\u0432\u044b\u0445 \u0434\u0430\u043d\u043d\u044b\u0445 \u044d\u0442\u0438 \u0432\u044b\u0432\u043e\u0434\u044b \u043e\u0442\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u043d\u0435 \u0431\u0443\u0434\u0443\u0442).<\/p>\n<p>\u0414\u043e\u0431\u0430\u0432\u0438\u043c \u0432\u044b\u0432\u043e\u0434 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c:<\/p>\n<pre><code class=\"python\">console = (console \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('console') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('console-output'))<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0441\u043e\u0445\u0440\u0430\u043d\u0435\u043d\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 Kafka, \u043f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u0438\u043c \u0442\u0430\u043a\u0436\u0435 JSON. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0432 Spark \u0435\u0441\u0442\u044c \u0444\u0443\u043d\u043a\u0446\u0438\u044f <code>to_json()<\/code>, \u043f\u0440\u0438\u043d\u0438\u043c\u0430\u044e\u0449\u0430\u044f <code>ColumnOrName<\/code>, \u0433\u0434\u0435 \u043c\u043e\u0436\u043d\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u043f\u043e\u043b\u0435, \u043a\u043e\u0442\u043e\u0440\u043e\u0435 \u0441\u043e\u0434\u0435\u0440\u0436\u0438\u0442 \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443:<\/p>\n<pre><code class=\"python\">df = df.withColumn('value', f.to_json('data'))  query = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('kafka-output') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('topic', 'output00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('checkpointLocation', '.\/.local\/checkpoint'))<\/code><\/pre>\n<p>\u0413\u043e\u0442\u043e\u0432\u043e!<\/p>\n<details class=\"spoiler\">\n<summary>\u041f\u043e\u043b\u043d\u044b\u0439 \u043a\u043e\u0434 \u043f\u0440\u0438\u043c\u0435\u0440\u0430<\/summary>\n<div class=\"spoiler__content\">\n<pre><code class=\"python\">def main(): \u00a0\u00a0\u00a0 from pyspark.sql import SparkSession \u00a0\u00a0\u00a0 from pyspark.sql import functions as f  \u00a0\u00a0\u00a0 from schema_example3 import schema  \u00a0\u00a0\u00a0 spark = (SparkSession \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .builder \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .appName('streaming-kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .getOrCreate()) \u00a0\u00a0\u00a0 spark.sparkContext.setLogLevel('WARN')  \u00a0\u00a0\u00a0 source = (spark \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .readStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('subscribe', 'input00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .load())  \u00a0\u00a0\u00a0 df = (source \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .selectExpr('CAST(value AS STRING)', 'offset'))  \u00a0\u00a0\u00a0 df = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .select(f.from_json('value', schema).alias('data'))) \u00a0\u00a0\u00a0 print(\"df schema\") \u00a0\u00a0\u00a0 df.printSchema()  \u00a0\u00a0\u00a0 df = df.where(df['data.age'] >= 18)  \u00a0\u00a0\u00a0 console = df.select('data.*') \u00a0\u00a0\u00a0 print(\"console schema\") \u00a0\u00a0\u00a0 console.printSchema()  \u00a0\u00a0\u00a0 console = (console \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('console') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('console-output'))  \u00a0\u00a0\u00a0 df = df.withColumn('value', f.to_json('data'))  \u00a0\u00a0\u00a0 query = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \u00a0.writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('kafka-output') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('topic', 'output00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('checkpointLocation', '.\/.local\/checkpoint'))  \u00a0\u00a0\u00a0 console.start() \u00a0\u00a0\u00a0 query.start().awaitTermination()   if __name__ == '__main__': \u00a0\u00a0\u00a0 main()<\/code><\/pre>\n<p>\u0421\u0445\u0435\u043c\u0430 \u0434\u0430\u043d\u043d\u044b\u0445:<\/p>\n<pre><code class=\"python\">from pyspark.sql import types as t  schema = t.StructType( \u00a0\u00a0\u00a0 [ \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 t.StructField('name', t.StringType(), True), \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 t.StructField('age', t.IntegerType(), True), \u00a0\u00a0\u00a0 ], )<\/code><\/pre>\n<\/p>\n<\/div>\n<\/details>\n<p>\u0417\u0430\u043f\u0438\u0448\u0435\u043c \u0442\u0440\u0438 JSON \u0441\u0442\u0440\u043e\u043a\u0438 \u0432 \u0442\u043e\u043f\u0438\u043a &#171;<strong>input00<\/strong>&#171;:<\/p>\n<pre><code>{\"name\":\"Ivan\",\"age\":15} {\"name\":\"Vladimir\",\"age\":33} {\"name\":\"Dmitry\",\"age\":24}<\/code><\/pre>\n<p>\u0418\u0437 \u043d\u0438\u0445 \u0442\u043e\u043b\u044c\u043a\u043e \u0434\u0432\u0435 \u0437\u0430\u043f\u0438\u0441\u0438 \u043f\u0440\u043e\u0445\u043e\u0434\u044f\u0442 \u0443\u0441\u043b\u043e\u0432\u0438\u0435 \u0444\u0438\u043b\u044c\u0442\u0440\u0430\u0446\u0438\u0438. \u0412\u044b\u0432\u043e\u0434 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u0435:<\/p>\n<pre><code>+--------+---+ |\u00a0\u00a0\u00a0 name|age| +--------+---+ |Vladimir| 33| |\u00a0 Dmitry| 24| +--------+---+<\/code><\/pre>\n<p>\u0421\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u0435 \u0442\u043e\u043f\u0438\u043a\u0430 &#171;<strong>output00<\/strong>&#171;:<\/p>\n<figure class=\"full-width\"><img loading=\"lazy\" decoding=\"async\" src=\"https:\/\/habrastorage.org\/r\/w1560\/getpro\/habr\/upload_files\/d16\/667\/7f7\/d166677f76030e87d42a59a86ae53f18.png\" alt=\"\u0421\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u0435 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u0433\u043e \u0442\u043e\u043f\u0438\u043a\u0430 Kafka\" title=\"\u0421\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u0435 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u0433\u043e \u0442\u043e\u043f\u0438\u043a\u0430 Kafka\" width=\"1353\" height=\"180\" data-src=\"https:\/\/habrastorage.org\/getpro\/habr\/upload_files\/d16\/667\/7f7\/d166677f76030e87d42a59a86ae53f18.png\"\/><figcaption>\u0421\u043e\u0441\u0442\u043e\u044f\u043d\u0438\u0435 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u0433\u043e \u0442\u043e\u043f\u0438\u043a\u0430 Kafka<\/figcaption><\/figure>\n<h2>6) \u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 AVRO<\/h2>\n<p>\u0420\u0430\u0431\u043e\u0442\u0430 \u0441 AVRO \u043d\u0435 \u0432\u0445\u043e\u0434\u0438\u0442 \u0432 \u043e\u0441\u043d\u043e\u0432\u043d\u043e\u0439 \u043c\u043e\u0434\u0443\u043b\u044c \u0440\u0430\u0431\u043e\u0442\u044b \u0441\u043e Spark. \u0414\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e, \u043a\u0430\u043a \u0438 \u0441 Kafka, \u043f\u043e\u0434\u0442\u044f\u0433\u0438\u0432\u0430\u0442\u044c \u0434\u043e\u043f\u043e\u043b\u043d\u0438\u0442\u0435\u043b\u044c\u043d\u044b\u0439 \u043f\u0430\u043a\u0435\u0442 \u043f\u0440\u0438 \u0437\u0430\u043f\u0443\u0441\u043a\u0435 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f:<\/p>\n<pre><code class=\"bash\">spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1,org.apache.spark:spark-avro_2.12:3.2.1 example4.py<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0444\u043e\u0440\u043c\u0430\u0442\u0430 AVRO \u0432 Spark \u0435\u0441\u0442\u044c \u0434\u0432\u0435 \u0444\u0443\u043d\u043a\u0446\u0438\u0438: <code>to_avro()<\/code> \u0438 <code>from_avro()<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u0445\u0440\u0430\u043d\u044f\u0442\u0441\u044f \u0432 \u043c\u043e\u0434\u0443\u043b\u0435 <code>pyspark.sql.avro<\/code>:<\/p>\n<pre><code class=\"python\">from pyspark.sql.avro import functions as fa<\/code><\/pre>\n<p>\u041f\u0435\u0440\u0435\u0432\u0435\u0434\u0451\u043c \u0441\u0442\u0440\u0443\u043a\u0442\u0443\u0440\u0443 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u043f\u0440\u043e\u0448\u043b\u043e\u0433\u043e \u043f\u0440\u0438\u043c\u0435\u0440\u0430 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442 AVRO \u0438 \u0441\u043e\u0445\u0440\u0430\u043d\u0438\u043c \u043a\u0430\u043a \u0444\u0430\u0439\u043b <code>*.avsc<\/code>:<\/p>\n<pre><code class=\"json\">{ \u00a0 \"type\": \"record\", \u00a0 \"name\": \"Person\", \u00a0 \"namespace\": \"example\", \u00a0 \"fields\": [ \u00a0\u00a0\u00a0 {\"name\": \"name\",\"type\": [\"string\", \"null\"]}, \u00a0\u00a0\u00a0 {\"name\": \"age\",\"type\": [\"long\", \"null\"]} \u00a0 ] }<\/code><\/pre>\n<p>\u0414\u043b\u044f \u043c\u0435\u0442\u043e\u0434\u0430 <code>from_avro()<\/code> \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043f\u0435\u0440\u0435\u0434\u0430\u0442\u044c \u043a\u043e\u043b\u043e\u043d\u043a\u0443 \u0438 \u0441\u0445\u0435\u043c\u0443 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 \u0441\u0442\u0440\u043e\u043a\u0438. \u041f\u043e\u043b\u0443\u0447\u0438\u0442\u044c \u0441\u0445\u0435\u043c\u0443 \u043c\u043e\u0436\u043d\u043e \u043e\u0431\u044b\u0447\u043d\u044b\u043c \u0441\u043f\u043e\u0441\u043e\u0431\u043e\u043c \u0438\u0437 \u0444\u0430\u0439\u043b\u0430:<\/p>\n<pre><code class=\"python\">schema = open('schema_example4.avsc', 'r').read()<\/code><\/pre>\n<p>\u041e\u0442\u043d\u043e\u0441\u0438\u0442\u0435\u043b\u044c\u043d\u043e \u043f\u0440\u0438\u043c\u0435\u0440\u0430 \u0441 JSON \u043e\u0441\u0442\u0430\u0451\u0442\u0441\u044f \u0442\u043e\u043b\u044c\u043a\u043e \u0437\u0430\u043c\u0435\u043d\u0438\u0442\u044c \u0444\u0443\u043d\u043a\u0446\u0438\u044e \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 \u043a\u043e\u043b\u043e\u043d\u043a\u0438, \u0433\u0434\u0435 \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 (\u0432 \u043e\u0442\u043b\u0438\u0447\u0438\u0438 \u043e\u0442 JSON, AVRO \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0432 \u0434\u0432\u043e\u0438\u0447\u043d\u043e\u043c \u0444\u043e\u0440\u043c\u0430\u0442\u0435, \u0430 \u0437\u043d\u0430\u0447\u0438\u0442 \u043d\u0430\u043c \u043d\u0435 \u043d\u0443\u0436\u043d\u043e \u0438\u0437\u043d\u0430\u0447\u0430\u043b\u044c\u043d\u043e \u043f\u0440\u0438\u0432\u043e\u0434\u0438\u0442\u044c \u043a\u043e\u043b\u043e\u043d\u043a\u0443 &#171;<strong>value<\/strong>&#187; \u043a \u0441\u0442\u0440\u043e\u043a\u043e\u0432\u043e\u043c\u0443 \u0442\u0438\u043f\u0443):<\/p>\n<pre><code class=\"python\">df = (source \u00a0\u00a0\u00a0\u00a0\u00a0 .select(fa.from_avro('value', schema).alias('data')))<\/code><\/pre>\n<p>\u041d\u0430 \u0432\u044b\u0445\u043e\u0434\u0435 <code>to_avro()<\/code> \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0441\u0445\u0435\u043c\u0443 \u043e\u043f\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u043e. \u041d\u043e \u043b\u0443\u0447\u0448\u0435 \u0432\u0441\u0451 \u0436\u0435 \u0435\u0451 \u0443\u043a\u0430\u0437\u0430\u0442\u044c, \u0438\u043d\u0430\u0447\u0435 \u043c\u043e\u0433\u0443\u0442 \u0432\u043e\u0437\u043d\u0438\u043a\u043d\u0443\u0442\u044c \u043f\u0440\u043e\u0431\u043b\u0435\u043c\u044b \u043f\u0440\u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 (\u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u0435\u0441\u043b\u0438 \u043f\u043e\u0440\u044f\u0434\u043e\u043a \u043a\u043e\u043b\u043e\u043d\u043e\u043a \u0431\u044b\u043b \u043d\u0430\u0440\u0443\u0448\u0435\u043d):<\/p>\n<pre><code class=\"python\">df = df.withColumn('value', fa.to_avro('data', schema))<\/code><\/pre>\n<p>\u0422\u0430\u043a\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c, \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u0438 \u0434\u0435\u0441\u0435\u0440\u0438\u0430\u043b\u0438\u0437\u043e\u0432\u0430\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 AVRO.<\/p>\n<details class=\"spoiler\">\n<summary>\u041f\u043e\u043b\u043d\u044b\u0439 \u043a\u043e\u0434 \u043f\u0440\u0438\u043c\u0435\u0440\u0430<\/summary>\n<div class=\"spoiler__content\">\n<pre><code class=\"python\">def main(): \u00a0\u00a0\u00a0 from pyspark.sql import SparkSession \u00a0\u00a0\u00a0 from pyspark.sql.avro import functions as fa  \u00a0\u00a0\u00a0 schema = open('schema_example4.avsc', 'r').read()  \u00a0\u00a0\u00a0 spark = (SparkSession \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .builder \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .appName('streaming-kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .getOrCreate()) \u00a0\u00a0\u00a0 spark.sparkContext.setLogLevel('WARN')  \u00a0\u00a0\u00a0 source = (spark \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .readStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('subscribe', 'input00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .load())  \u00a0\u00a0\u00a0 df = (source \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .select(fa.from_avro('value', schema).alias('data')))  \u00a0\u00a0\u00a0 print(\"df schema\") \u00a0\u00a0\u00a0 df.printSchema()  \u00a0\u00a0\u00a0 df = df.where(df['data.age'] >= 18)  \u00a0\u00a0\u00a0 console = df.select('data.*') \u00a0\u00a0\u00a0 print(\"console schema\") \u00a0\u00a0\u00a0 console.printSchema()  \u00a0\u00a0\u00a0 console = (console \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('console') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('console-output'))  \u00a0\u00a0\u00a0 df = df.withColumn('value', fa.to_avro('data', schema))  \u00a0\u00a0\u00a0 query = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('kafka-output') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('topic', 'output00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('checkpointLocation', '.\/.local\/checkpoint'))  \u00a0\u00a0\u00a0 console.start() \u00a0\u00a0\u00a0 query.start().awaitTermination()   if __name__ == '__main__': \u00a0\u00a0\u00a0 main()<\/code><\/pre>\n<p>\u0421\u0445\u0435\u043c\u0430 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 AVRO:<\/p>\n<pre><code class=\"json\">{ \u00a0 \"type\": \"record\", \u00a0 \"name\": \"Person\", \u00a0 \"namespace\": \"com.neoflex.example\", \u00a0 \"fields\": [ \u00a0\u00a0\u00a0 {\"name\": \"name\",\"type\": [\"string\", \"null\"]}, \u00a0\u00a0\u00a0 {\"name\": \"age\",\"type\": [\"long\", \"null\"]} \u00a0 ] }<\/code><\/pre>\n<\/p>\n<\/div>\n<\/details>\n<p>\u0414\u043b\u044f \u043f\u0443\u0431\u043b\u0438\u043a\u0430\u0446\u0438\u0438 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 AVRO \u0432 Kafka \u0431\u044b\u043b \u043d\u0430\u043f\u0438\u0441\u0430\u043d \u0438 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d \u043a \u043f\u0440\u0438\u043c\u0435\u0440\u0430\u043c Producer. \u0412\u0441\u0435 \u043f\u0440\u0438\u043c\u0435\u0440\u044b \u0438\u0437 \u0441\u0442\u0430\u0442\u044c\u0438 \u0434\u043e\u0441\u0442\u0443\u043f\u043d\u044b \u0432 <a href=\"https:\/\/github.com\/dedkot01\/article-examples-pyspark-streaming-with-kafka\">\u0440\u0435\u043f\u043e\u0437\u0438\u0442\u043e\u0440\u0438\u0438<\/a>.<\/p>\n<h2>\u0417\u0430\u043a\u043b\u044e\u0447\u0435\u043d\u0438\u0435<\/h2>\n<p>\u0426\u0435\u043b\u044c \u0441\u0442\u0430\u0442\u044c\u0438 \u2013 \u043d\u0430\u0443\u0447\u0438\u0442\u044c\u0441\u044f\u00a0\u043f\u0438\u0441\u0430\u0442\u044c \u043f\u0440\u043e\u0441\u0442\u0435\u0439\u0448\u0438\u0435 \u043f\u0440\u043e\u0446\u0435\u0441\u0441\u044b \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u043e\u0439 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0441 \u043f\u0440\u0438\u043c\u0435\u043d\u0435\u043d\u0438\u0435\u043c \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u0430 PySpark, \u0432\u0437\u0430\u0438\u043c\u043e\u0434\u0435\u0439\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c \u0441 Kafka \u0438 \u0447\u0438\u0442\u0430\u0442\u044c\/\u043f\u0438\u0441\u0430\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0430\u0445 JSON \u0438 AVRO. Spark \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u0441\u0438\u043d\u0442\u0430\u043a\u0441\u0438\u0441 SQL, \u0431\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f \u0447\u0435\u043c\u0443 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u044b \u043f\u043e\u043b\u0443\u0447\u0430\u044e\u0442\u0441\u044f \u0443\u0434\u043e\u0431\u043e\u0447\u0438\u0442\u0430\u0435\u043c\u044b \u0437\u0430 \u0441\u0447\u0451\u0442 \u043e\u0442\u0441\u0443\u0442\u0441\u0442\u0432\u0438\u044f \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e\u0441\u0442\u0438 \u0438\u0437\u0443\u0447\u0430\u0442\u044c \u0441\u0438\u043d\u0442\u0430\u043a\u0441\u0438\u0441 \u043e\u0447\u0435\u0440\u0435\u0434\u043d\u043e\u0433\u043e \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u0430, \u0432\u0435\u0434\u044c SQL \u0434\u043e\u0441\u0442\u0430\u0442\u043e\u0447\u043d\u043e \u0448\u0438\u0440\u043e\u043a\u043e \u0438\u0437\u0432\u0435\u0441\u0442\u0435\u043d.<\/p>\n<p>\u0422\u0430\u043a\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c, \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u0431\u0440\u043e\u043a\u0435\u0440 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f Kafka \u0438 PySpark Structured Streaming, \u0432\u044b \u0441\u043c\u043e\u0436\u0435\u0442\u0435 \u043e\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c \u0441\u0432\u043e\u0438 \u0441\u0438\u0441\u0442\u0435\u043c\u044b \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u043b\u044e\u0431\u043e\u0439 \u0441\u043b\u043e\u0436\u043d\u043e\u0441\u0442\u0438 \u0438 \u043c\u0430\u0441\u0448\u0442\u0430\u0431\u0438\u0440\u0443\u0435\u043c\u043e\u0441\u0442\u0438.<\/p>\n<p>\u0410\u0432\u0442\u043e\u0440 \u0441\u0442\u0430\u0442\u044c\u0438: \u0414\u043c\u0438\u0442\u0440\u0438\u0439 \u0416\u0434\u0430\u043d\u043e\u0432, \u0441\u043f\u0435\u0446\u0438\u0430\u043b\u0438\u0441\u0442 \u0431\u0438\u0437\u043d\u0435\u0441-\u043d\u0430\u043f\u0440\u0430\u0432\u043b\u0435\u043d\u0438\u044f Fast Data \u043a\u043e\u043c\u043f\u0430\u043d\u0438\u0438 Neoflex.<\/p>\n<\/p>\n<\/div>\n<\/div>\n<\/div>\n<div class=\"v-portal\" style=\"display:none;\"><\/div>\n<\/div>\n<p> <!----> <!----><br \/> \u0441\u0441\u044b\u043b\u043a\u0430 \u043d\u0430 \u043e\u0440\u0438\u0433\u0438\u043d\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 <a href=\"https:\/\/habr.com\/ru\/company\/neoflex\/blog\/674944\/\"> https:\/\/habr.com\/ru\/company\/neoflex\/blog\/674944\/<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"<div><\/div>\n<div id=\"post-content-body\">\n<div>\n<div class=\"article-formatted-body article-formatted-body article-formatted-body_version-2\">\n<div xmlns=\"http:\/\/www.w3.org\/1999\/xhtml\">\n<figure class=\"full-width\"><figcaption><\/figcaption><\/figure>\n<p>\u0414\u0430\u043d\u043d\u0430\u044f \u0441\u0442\u0430\u0442\u044c\u044f \u043e\u0431\u043e\u0431\u0449\u0430\u0435\u0442 \u0431\u0430\u0437\u043e\u0432\u044b\u0435 \u0448\u0430\u0433\u0438 \u043f\u043e \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043a\u0435 \u0438 \u043d\u0430\u0447\u0430\u043b\u0443 \u0440\u0430\u0431\u043e\u0442\u044b \u0441 PySpark Structured Streaming \u043f\u0440\u0438 \u0443\u0447\u0430\u0441\u0442\u0438\u0438 \u0431\u0440\u043e\u043a\u0435\u0440\u0430 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 Kafka. \u041f\u0440\u0435\u0434\u043f\u043e\u043b\u0430\u0433\u0430\u0435\u0442\u0441\u044f, \u0447\u0442\u043e \u0447\u0438\u0442\u0430\u0442\u0435\u043b\u044c \u0443\u0436\u0435 \u0437\u043d\u0430\u043a\u043e\u043c \u0441 \u044f\u0437\u044b\u043a\u043e\u043c \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0438\u0440\u043e\u0432\u0430\u043d\u0438\u044f Python \u0438 \u0441\u0435\u0440\u0432\u0438\u0441\u043e\u043c Kafka.<\/p>\n<p>\u041f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 PySpark Structured Streaming \u043c\u043e\u0436\u043d\u043e \u0431\u044b\u0441\u0442\u0440\u043e \u0440\u0430\u0437\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u043c\u0430\u0441\u0448\u0442\u0430\u0431\u0438\u0440\u0443\u0435\u043c\u044b\u0435 \u0441\u0435\u0440\u0432\u0438\u0441\u044b \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438. \u0422\u0430\u043a\u043e\u0439 \u043f\u043e\u0434\u0445\u043e\u0434 \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0432 \u043a\u043e\u0440\u043e\u0442\u043a\u0438\u0435 \u0441\u0440\u043e\u043a\u0438 \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u0432\u044b\u0433\u043e\u0434\u043d\u043e\u0435 \u043f\u0440\u0435\u0434\u043b\u043e\u0436\u0435\u043d\u0438\u0435 \u043a\u043b\u0438\u0435\u043d\u0442\u0443, \u0432\u043e\u0432\u0440\u0435\u043c\u044f \u0437\u0430\u043c\u0435\u0442\u0438\u0442\u044c \u0430\u043d\u043e\u043c\u0430\u043b\u0438\u044e \u0432 \u0441\u0438\u0441\u0442\u0435\u043c\u0435 \u0438\u043b\u0438 \u0436\u0435 \u043e\u0442\u043e\u0431\u0440\u0430\u0436\u0430\u0442\u044c \u0430\u043a\u0442\u0443\u0430\u043b\u044c\u043d\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435. \u041c\u0430\u0441\u0448\u0442\u0430\u0431\u0438\u0440\u0443\u0435\u043c\u043e\u0441\u0442\u044c \u043e\u0431\u0435\u0441\u043f\u0435\u0447\u0438\u0432\u0430\u0435\u0442\u0441\u044f \u0444\u0440\u0435\u0439\u043c\u0432\u043e\u0440\u043a\u043e\u043c Spark. \u041c\u043e\u0434\u0443\u043b\u044c Structured Streaming \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u0435\u0442 \u0440\u0430\u0437\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u044b \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438, \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u0441\u0438\u043d\u0442\u0430\u043a\u0441\u0438\u0441 SQL.<\/p>\n<p>\u0417\u0434\u0435\u0441\u044c \u0431\u0443\u0434\u0443\u0442 \u0440\u0430\u0441\u0441\u043c\u043e\u0442\u0440\u0435\u043d\u044b \u0442\u043e\u043b\u044c\u043a\u043e \u0441\u0430\u043c\u044b\u0435 \u043f\u0440\u043e\u0441\u0442\u044b\u0435 \u043f\u0440\u0438\u043c\u0435\u0440\u044b \u0440\u0430\u0431\u043e\u0442\u044b \u0441 Kafka \u043f\u043e\u0441\u0440\u0435\u0434\u0441\u0442\u0432\u043e\u043c PySpark: \u0447\u0442\u0435\u043d\u0438\u0435\/\u0437\u0430\u043f\u0438\u0441\u044c \u0432 Kafka, \u0430 \u0442\u0430\u043a\u0436\u0435 \u0440\u0430\u0437\u0431\u043e\u0440 \u0438 \u0441\u043e\u0445\u0440\u0430\u043d\u0435\u043d\u0438\u0435 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0445 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 JSON \u0438 AVRO. \u0414\u043b\u044f \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e\u0433\u043e \u043e\u0437\u043d\u0430\u043a\u043e\u043c\u043b\u0435\u043d\u0438\u044f \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0441\u0442\u0430\u0442\u044c\u0438 \u043d\u0430 \u043e\u0444\u0438\u0446\u0438\u0430\u043b\u044c\u043d\u043e\u043c \u0441\u0430\u0439\u0442\u0435:<\/p>\n<ul>\n<li>\n<p><a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/structured-streaming-programming-guide.html\">Spark Structured Streaming<\/a><\/p>\n<\/li>\n<li>\n<p><a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/structured-streaming-kafka-integration.html\">Spark Structured Streaming Kafka Integration<\/a><\/p>\n<\/li>\n<\/ul>\n<h2>\u0422\u0440\u0435\u0431\u043e\u0432\u0430\u043d\u0438\u044f \u043a \u043e\u043a\u0440\u0443\u0436\u0435\u043d\u0438\u044e<\/h2>\n<p>\u0412\u0441\u0435 \u043f\u0440\u0438\u043c\u0435\u0440\u044b \u0431\u044b\u043b\u0438 \u043e\u0442\u043b\u0430\u0436\u0435\u043d\u044b \u043d\u0430 \u043f\u043e\u0434\u0441\u0438\u0441\u0442\u0435\u043c\u0435 Ubuntu 20.04, Python 3.10.1, Spark 3.2.1, Kafka 3, OpenJDK 11.0.11 (\u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u0430 \u0434\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b Spark).<\/p>\n<h2>1.1) \u041f\u043e\u0434\u0433\u043e\u0442\u043e\u0432\u043a\u0430 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u0441\u0440\u0435\u0434\u044b<\/h2>\n<p>\u0421\u043e\u0437\u0434\u0430\u0451\u043c \u043f\u0443\u0441\u0442\u0443\u044e \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u044e \u0441 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435\u043c \u043f\u0440\u043e\u0435\u043a\u0442\u0430, \u0433\u0434\u0435 \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0438\u0440\u0443\u0435\u043c \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u0443\u044e \u0441\u0440\u0435\u0434\u0443 (\u043a\u043e\u043c\u0430\u043d\u0434\u0430 \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435):<\/p>\n<pre><code class=\"bash\">python -m venv .venv<\/code><\/pre>\n<p>\u0410\u043a\u0442\u0438\u0432\u0438\u0440\u0443\u0435\u043c \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435 \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u0441\u043b\u0435\u0434\u0443\u044e\u0449\u0435\u0439 \u043a\u043e\u043c\u0430\u043d\u0434\u044b (\u0438\u0437 \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u0438, \u0433\u0434\u0435 \u0431\u044b\u043b\u0430 \u0441\u043e\u0437\u0434\u0430\u043d\u0430 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u0430\u044f \u0441\u0440\u0435\u0434\u0430):<\/p>\n<pre><code class=\"bash\">source .venv\/bin\/activate<\/code><\/pre>\n<p>\u0423\u0441\u0442\u0430\u043d\u0430\u0432\u043b\u0438\u0432\u0430\u0435\u043c PySpark:<\/p>\n<pre><code class=\"bash\">pip install wheel pyspark<\/code><\/pre>\n<p>\u0427\u0442\u043e\u0431\u044b \u043f\u0440\u043e\u0432\u0435\u0440\u0438\u0442\u044c, \u0447\u0442\u043e \u0432\u0441\u0451 \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u043e\u0441\u044c \u043a\u043e\u0440\u0440\u0435\u043a\u0442\u043d\u043e, \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0438\u043d\u0442\u0435\u0440\u043f\u0440\u0435\u0442\u0430\u0442\u043e\u0440 PySpark:<\/p>\n<pre><code class=\"bash\">pyspark<\/code><\/pre>\n<p>\u0414\u043e\u043b\u0436\u0435\u043d \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c\u0441\u044f \u0438\u043d\u0442\u0435\u0440\u043f\u0440\u0435\u0442\u0430\u0442\u043e\u0440 Python, \u0433\u0434\u0435 \u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d Spark.<\/p>\n<figure class=\"bordered\"><figcaption>\u041f\u0440\u0438\u0432\u0435\u0442\u0441\u0442\u0432\u0443\u044e\u0449\u0435\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 Spark<\/figcaption><\/figure>\n<h2>1.2) \u0417\u0430\u043f\u0443\u0441\u043a \u0441\u0435\u0440\u0432\u0438\u0441\u0430 Kafka<\/h2>\n<p>\u0414\u043b\u044f \u043f\u0440\u0438\u043c\u0435\u0440\u0430 Kafka \u0431\u0443\u0434\u0435\u0442 \u0437\u0430\u043f\u0443\u0449\u0435\u043d\u0430 \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e. \u042d\u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0441\u0434\u0435\u043b\u0430\u0442\u044c, \u0432\u044b\u043f\u043e\u043b\u043d\u0438\u0432 \u0434\u0435\u0439\u0441\u0442\u0432\u0438\u044f \u0438\u0437 <a href=\"https:\/\/kafka.apache.org\/quickstart\">\u043e\u0444\u0438\u0446\u0438\u0430\u043b\u044c\u043d\u043e\u0439 \u0438\u043d\u0441\u0442\u0440\u0443\u043a\u0446\u0438\u0438<\/a>. \u0415\u0441\u043b\u0438 \u0443\u043c\u0435\u0435\u0442\u0435 \u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c\u0441\u044f Docker, \u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c \u0435\u0451 \u043f\u0440\u0438 \u043f\u043e\u043c\u043e\u0449\u0438 \u0434\u0430\u043d\u043d\u043e\u0433\u043e <a href=\"https:\/\/github.com\/bitnami\/bitnami-docker-kafka\/blob\/master\/docker-compose.yml\">docker-compose<\/a> \u0438\u043b\u0438 \u0436\u0435 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u043e\u0432\u0430\u0442\u044c \u043b\u044e\u0431\u043e\u0439 \u0443\u0434\u043e\u0431\u043d\u044b\u0439 \u0434\u043b\u044f \u0432\u0430\u0441 \u043e\u0431\u0440\u0430\u0437.<\/p>\n<h2>2) \u0427\u0442\u0435\u043d\u0438\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439 \u0438\u0437 \u0442\u043e\u043f\u0438\u043a\u0430 Kafka<\/h2>\n<p>\u0421\u043e\u0437\u0434\u0430\u0451\u043c \u043b\u044e\u0431\u043e\u0439 \u0444\u0430\u0439\u043b <code>*.py<\/code>. \u0414\u043b\u044f \u043f\u0440\u0438\u043c\u0435\u0440\u0430 \u044d\u0442\u043e \u0431\u0443\u0434\u0435\u0442 <code>example1.py<\/code>.<\/p>\n<p>\u0414\u043b\u044f \u0440\u0430\u0431\u043e\u0442\u044b \u0441\u043e Spark \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043e\u0431\u044a\u0435\u043a\u0442 <code>SparkSession<\/code>:<\/p>\n<pre><code class=\"python\">from pyspark.sql import SparkSession  spark = (SparkSession \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .builder \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .appName('quickstart-streaming-kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .getOrCreate()) spark.sparkContext.setLogLevel('WARN')<\/code><\/pre>\n<p>\u041f\u0440\u0438 \u0438\u043d\u0438\u0446\u0438\u0430\u043b\u0438\u0437\u0430\u0446\u0438\u0438 <code>SparkSession<\/code> \u0432\u044b \u043c\u043e\u0436\u0435\u0442\u0435 \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0440\u0430\u0437\u043b\u0438\u0447\u043d\u044b\u0435 \u0438\u043c\u0435\u043d\u0430 \u0432 \u043c\u0435\u0442\u043e\u0434\u0435 <code>appName()<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u00a0\u043f\u043e\u0437\u0432\u043e\u043b\u044f\u044e\u0442 \u043f\u043e\u0437\u0436\u0435 \u043e\u043f\u043e\u0437\u043d\u0430\u0432\u0430\u0442\u044c \u0441\u0432\u043e\u0438 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u044f (\u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u043f\u0440\u0438 \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433\u0435 \u0432\u0441\u0435\u0445 \u0437\u0430\u043f\u0443\u0449\u0435\u043d\u043d\u044b\u0445 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 Spark \u043d\u0430 \u043a\u043b\u0430\u0441\u0442\u0435\u0440\u0435).<\/p>\n<p>\u041f\u043e\u0441\u043b\u0435\u0434\u043d\u044f\u044f \u0441\u0442\u0440\u043e\u043a\u0430 \u043e\u043f\u0446\u0438\u043e\u043d\u0430\u043b\u044c\u043d\u0430, \u043e\u043d\u0430 \u0432\u043b\u0438\u044f\u0435\u0442 \u043d\u0430 \u0432\u044b\u0432\u043e\u0434 \u043b\u043e\u0433\u043e\u0432 \u0440\u0430\u0431\u043e\u0442\u044b Spark. \u0423\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u0432 \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435 <strong>&#171;WARN&#187;<\/strong>, \u043d\u0435 \u0431\u0443\u0434\u0443\u0442 \u0432\u044b\u0432\u043e\u0434\u0438\u0442\u044c\u0441\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u043e\u043d\u043d\u044b\u0435 \u043b\u043e\u0433\u0438.<\/p>\n<p>\u0423\u043a\u0430\u0436\u0435\u043c \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a \u0434\u0430\u043d\u043d\u044b\u0445 \u2013 \u0442\u043e\u043f\u0438\u043a Kafka:<\/p>\n<pre><code class=\"python\">source = (spark \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .readStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('subscribe', 'input00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .load())<\/code><\/pre>\n<p>\u0418\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043e\u0431\u044a\u0435\u043a\u0442 <code>SparkSession<\/code>, \u043c\u043e\u0436\u043d\u043e \u0443\u043a\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a \u0434\u0430\u043d\u043d\u044b\u0445, \u0431\u043b\u0430\u0433\u043e\u0434\u0430\u0440\u044f \u0447\u0435\u043c\u0443 \u043f\u043e\u043b\u0443\u0447\u0438\u043c \u043e\u0431\u044a\u0435\u043a\u0442 <code>DataFrame<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0438 \u0431\u0443\u0434\u0435\u0442 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u044f\u0442\u044c \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c\u044b\u0435 \u0434\u0430\u043d\u043d\u044b\u0435. \u0418\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0438 \u0432 Spark \u0434\u0435\u043b\u044f\u0442\u0441\u044f \u043d\u0430 \u0434\u0432\u0430 \u0442\u0438\u043f\u0430:<\/p>\n<ul>\n<li>\n<p><strong>batch<\/strong> \u2013 \u0440\u0430\u0437\u043e\u0432\u0430\u044f \u0440\u0430\u0431\u043e\u0442\u0430, \u043a\u043e\u0433\u0434\u0430 \u0432\u0430\u043c \u0437\u0430\u0440\u0430\u043d\u0435\u0435 \u0438\u0437\u0432\u0435\u0441\u0442\u0435\u043d \u043e\u0431\u044a\u0451\u043c \u0434\u0430\u043d\u043d\u044b\u0445 (\u0434\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0441\u0432\u043e\u0439\u0441\u0442\u0432\u043e <code>read<\/code> \u0443 <code>SparkSession<\/code>);<\/p>\n<\/li>\n<li>\n<p><strong>stream<\/strong> \u2013 \u043f\u043e\u0442\u043e\u043a\u043e\u0432\u0430\u044f \u0440\u0430\u0431\u043e\u0442\u0430, \u043a\u043e\u0433\u0434\u0430 \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0435 \u0437\u0430\u043f\u0443\u0449\u0435\u043d\u043e \u0438 \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0435\u0442 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0435 \u0434\u0430\u043d\u043d\u044b\u0435 (\u0434\u043b\u044f \u044d\u0442\u043e\u0433\u043e \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0441\u0432\u043e\u0439\u0441\u0442\u0432\u043e <code>readStream<\/code>) \u0432 \u0440\u0435\u0430\u043b\u044c\u043d\u043e\u043c \u0432\u0440\u0435\u043c\u0435\u043d\u0438. \u0412 \u0434\u0430\u043d\u043d\u043e\u043c \u0440\u0435\u0436\u0438\u043c\u0435 Spark \u0441 \u043d\u0435\u043a\u043e\u0442\u043e\u0440\u043e\u0439 \u043f\u0435\u0440\u0438\u043e\u0434\u0438\u0447\u043d\u043e\u0441\u0442\u044c\u044e \u0444\u043e\u0440\u043c\u0438\u0440\u0443\u0435\u0442 batch \u0438\u0437 \u043f\u043e\u0441\u0442\u0443\u043f\u0430\u044e\u0449\u0438\u0445 \u0434\u0430\u043d\u043d\u044b\u0445.<\/p>\n<\/li>\n<\/ul>\n<p>\u0427\u0442\u043e\u0431\u044b \u0441\u0447\u0438\u0442\u044b\u0432\u0430\u0442\u044c \u0438\u0437 Kafka, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432 \u0444\u043e\u0440\u043c\u0430\u0442\u0435 <code>format()<\/code> \u043f\u0435\u0440\u0435\u0434\u0430\u0442\u044c \u0441\u0442\u0440\u043e\u043a\u0443 &#171;<strong>kafka<\/strong>&#171;. \u0421\u043f\u0438\u0441\u043e\u043a \u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u044b\u0445 \u043e\u043f\u0446\u0438\u0439 \u043f\u0440\u0438 \u0440\u0430\u0431\u043e\u0442\u0435 \u0441 \u0434\u0430\u043d\u043d\u044b\u043c \u0444\u043e\u0440\u043c\u0430\u0442\u043e\u043c:<\/p>\n<ul>\n<li>\n<p><strong>kafka.bootstrap.servers<\/strong> \u2013 \u0430\u0434\u0440\u0435\u0441 \u0441\u0435\u0440\u0432\u0435\u0440\u0430 \u0431\u0440\u043e\u043a\u0435\u0440\u0430 (\u043c\u043e\u0436\u043d\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u2013 \u0447\u0435\u0440\u0435\u0437 \u0437\u0430\u043f\u044f\u0442\u0443\u044e \u0432\u043d\u0443\u0442\u0440\u0438 \u0441\u0442\u0440\u043e\u043a\u0438);<\/p>\n<\/li>\n<li>\n<p><strong>subscribe<\/strong> \u2013 \u0442\u043e\u043f\u0438\u043a, \u043e\u0442\u043a\u0443\u0434\u0430 \u0431\u0443\u0434\u0443\u0442 \u0441\u0447\u0438\u0442\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u043d\u043e\u0432\u044b\u0435 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f (\u043c\u043e\u0436\u043d\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u2013 \u0447\u0435\u0440\u0435\u0437 \u0437\u0430\u043f\u044f\u0442\u0443\u044e \u0432\u043d\u0443\u0442\u0440\u0438 \u0441\u0442\u0440\u043e\u043a\u0438).<\/p>\n<\/li>\n<\/ul>\n<p>\u041f\u043e\u0441\u043c\u043e\u0442\u0440\u0438\u043c \u0441\u0445\u0435\u043c\u0443 \u043f\u043e\u043b\u0443\u0447\u0430\u0435\u043c\u044b\u0445 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0439, \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043c\u0435\u0442\u043e\u0434 <code>printSchema()<\/code> \u0443 \u043f\u043e\u043b\u0443\u0447\u0438\u0432\u0448\u0435\u0433\u043e\u0441\u044f <code>DataFrame<\/code>:<\/p>\n<pre><code class=\"python\">source.printSchema()<\/code><\/pre>\n<p>\u0412\u044b\u0432\u043e\u0434 \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435:<\/p>\n<pre><code>root \u00a0|-- key: binary (nullable = true) \u00a0|-- value: binary (nullable = true) \u00a0|-- topic: string (nullable = true) \u00a0|-- partition: integer (nullable = true) \u00a0|-- offset: long (nullable = true) \u00a0|-- timestamp: timestamp (nullable = true) \u00a0|-- timestampType: integer (nullable = true)<\/code><\/pre>\n<p>\u041a\u0440\u0430\u0442\u043a\u0430\u044f \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043f\u043e \u043a\u0430\u0436\u0434\u043e\u043c\u0443 \u043f\u043e\u043b\u044e:<\/p>\n<ol>\n<li>\n<p><strong>key<\/strong> \u2013 \u043a\u043b\u044e\u0447, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f \u0432 Kafka \u0434\u043b\u044f \u043e\u043f\u0440\u0435\u0434\u0435\u043b\u0435\u043d\u0438\u044f \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u0438, \u0432 \u043a\u043e\u0442\u043e\u0440\u0443\u044e \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u043b\u043e \u043f\u043e\u043f\u0430\u0441\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435;<\/p>\n<\/li>\n<li>\n<p><strong>value<\/strong> \u2013 \u043d\u0435\u043f\u043e\u0441\u0440\u0435\u0434\u0441\u0442\u0432\u0435\u043d\u043d\u043e \u0441\u0430\u043c\u0438 \u0434\u0430\u043d\u043d\u044b\u0435, \u043a\u043e\u0442\u043e\u0440\u044b\u0435 \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043e\u0431\u0440\u0430\u0431\u0430\u0442\u044b\u0432\u0430\u0442\u044c;<\/p>\n<\/li>\n<li>\n<p><strong>topic<\/strong> \u2013 \u043d\u0430\u0437\u0432\u0430\u043d\u0438\u0435 \u0442\u043e\u043f\u0438\u043a\u0430, \u043e\u0442\u043a\u0443\u0434\u0430 \u0431\u044b\u043b\u043e \u043f\u043e\u043b\u0443\u0447\u0435\u043d\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 (\u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0438 \u0432 Spark \u043f\u043e\u0437\u0432\u043e\u043b\u044f\u044e\u0442 \u043f\u043e\u0434\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c\u0441\u044f \u0441\u0440\u0430\u0437\u0443 \u043d\u0430 \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0442\u043e\u043f\u0438\u043a\u043e\u0432, \u0438\u0445 \u043c\u043e\u0436\u043d\u043e \u0438\u0434\u0435\u043d\u0442\u0438\u0444\u0438\u0446\u0438\u0440\u043e\u0432\u0430\u0442\u044c \u043f\u043e \u0434\u0430\u043d\u043d\u043e\u043c\u0443 \u043f\u043e\u043b\u044e);<\/p>\n<\/li>\n<li>\n<p><strong>partition<\/strong> \u0438 <strong>offset<\/strong> \u2013 \u043f\u0430\u0440\u0442\u0438\u0446\u0438\u044f \u0438 \u0441\u043c\u0435\u0449\u0435\u043d\u0438\u0435 \u0441\u043b\u0443\u0436\u0430\u0442 \u0443\u043a\u0430\u0437\u0430\u0442\u0435\u043b\u0435\u043c, \u0433\u0434\u0435 \u0438\u043c\u0435\u043d\u043d\u043e \u0432 \u0442\u043e\u043f\u0438\u043a\u0435 \u0445\u0440\u0430\u043d\u0438\u0442\u0441\u044f \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435;<\/p>\n<\/li>\n<li>\n<p><strong>timestamp<\/strong> \u0438 <strong>timestampType<\/strong> \u2013 \u0438\u043d\u0444\u043e\u0440\u043c\u0430\u0446\u0438\u044f \u043e \u0432\u0440\u0435\u043c\u0435\u043d\u0438, \u043a\u043e\u0433\u0434\u0430 \u043f\u043e\u0441\u0442\u0443\u043f\u0438\u043b\u043e \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a.<\/p>\n<\/li>\n<\/ol>\n<p>\u041c\u043e\u0436\u043d\u043e \u0437\u0430\u043c\u0435\u0442\u0438\u0442\u044c, \u0447\u0442\u043e \u043f\u043e\u043b\u044f &#171;<strong>key<\/strong>&#187; \u0438 &#171;<strong>value<\/strong>&#187; \u0438\u043c\u0435\u044e\u0442 \u0434\u0432\u043e\u0438\u0447\u043d\u044b\u0439 \u0444\u043e\u0440\u043c\u0430\u0442. \u0418 \u0435\u0441\u043b\u0438 \u043f\u043e\u0441\u0442\u0443\u043f\u0438\u0442 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0441\u043e \u0437\u043d\u0430\u0447\u0435\u043d\u0438\u0435\u043c &#171;<strong>Hello, World!<\/strong>&#171;, \u0442\u043e \u043e\u043d\u043e \u0431\u0443\u0434\u0435\u0442 \u043f\u0440\u0435\u0434\u0441\u0442\u0430\u0432\u043b\u0435\u043d\u043e \u043c\u0430\u0441\u0441\u0438\u0432\u043e\u043c \u0431\u0430\u0439\u0442\u043e\u0432:<\/p>\n<pre><code>------------------------------------------- Batch: 1 ------------------------------------------- +----+--------------------+-------+---------+------+--------------------+-------------+ | key|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 value|\u00a0 topic|partition|offset|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 timestamp|timestampType| +----+--------------------+-------+---------+------+--------------------+-------------+ |[30]|[48 65 6C 6C 6F 2...|input00|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 0|\u00a0\u00a0\u00a0\u00a0 3|2022-03-11 08:28:...|\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 0| +----+--------------------+-------+---------+------+--------------------+-------------+<\/code><\/pre>\n<p>\u0412 Spark \u043c\u043e\u0436\u043d\u043e \u0432\u044b\u043f\u043e\u043b\u043d\u044f\u0442\u044c \u0432\u044b\u0431\u043e\u0440\u043a\u0443 \u0441 \u043f\u0440\u0438\u043c\u0435\u043d\u0435\u043d\u0438\u0435\u043c \u0440\u0430\u0437\u043b\u0438\u0447\u043d\u044b\u0445 \u0432\u044b\u0440\u0430\u0436\u0435\u043d\u0438\u0439, \u043d\u0430\u043f\u0440\u0438\u043c\u0435\u0440, \u043f\u0440\u0438\u0432\u0435\u0434\u0435\u043d\u0438\u0435 \u0442\u0438\u043f\u043e\u0432. \u0418\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u044f \u043c\u0435\u0442\u043e\u0434 <code>selectExpr()<\/code>, \u0433\u0434\u0435 \u043c\u043e\u0436\u043d\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0432\u044b\u0432\u043e\u0434\u0438\u043c\u044b\u0435 \u043f\u043e\u043b\u044f \u0432 <code>DataFrame<\/code>, \u0432\u044b\u0432\u0435\u0434\u0435\u043c \u0438 \u043f\u0440\u0435\u043e\u0431\u0440\u0430\u0437\u0443\u0435\u043c \u043f\u043e\u043b\u0435 &#171;<strong>value<\/strong>&#187; \u043a \u0441\u0442\u0440\u043e\u043a\u0435, \u0430 \u0442\u0430\u043a\u0436\u0435 \u0432\u044b\u0432\u0435\u0434\u0435\u043c &#171;<strong>offset<\/strong>&#171;:<\/p>\n<pre><code class=\"python\">df = (source \u00a0\u00a0\u00a0\u00a0\u00a0 .selectExpr('CAST(value AS STRING)', 'offset'))<\/code><\/pre>\n<p>\u041d\u0430\u043f\u0438\u0448\u0435\u043c \u043f\u0440\u043e\u0441\u0442\u043e\u0439 \u0432\u044b\u0432\u043e\u0434 \u0434\u0430\u043d\u043d\u044b\u0445 \u0438\u0437 \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0430 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c:<\/p>\n<pre><code class=\"python\">console = (df            .writeStream            .format('console'))<\/code><\/pre>\n<p>\u0423 \u043b\u044e\u0431\u043e\u0433\u043e <code>DataFrame<\/code> \u043c\u043e\u0436\u043d\u043e \u0432\u044b\u0437\u0432\u0430\u0442\u044c \u0441\u0432\u043e\u0439\u0441\u0442\u0432\u043e <code>writeStream<\/code>, \u043f\u043e\u0441\u043b\u0435 \u0447\u0435\u0433\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0444\u043e\u0440\u043c\u0430\u0442 \u0432\u044b\u0432\u043e\u0434\u0430. \u00a0\u0412 \u043d\u0430\u0448\u0435\u043c \u0441\u043b\u0443\u0447\u0430\u0435 \u044d\u0442\u043e \u0432\u044b\u0432\u043e\u0434 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c (\u043f\u043e\u043b\u043d\u044b\u0439 \u0441\u043f\u0438\u0441\u043e\u043a \u043f\u043e\u0434\u0434\u0435\u0440\u0436\u0438\u0432\u0430\u0435\u043c\u044b\u0445 \u0444\u043e\u0440\u043c\u0430\u0442\u043e\u0432 \u0434\u043e\u0441\u0442\u0443\u043f\u0435\u043d \u0432 <a href=\"https:\/\/spark.apache.org\/docs\/latest\/structured-streaming-programming-guide.html#output-sinks\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>). \u0422\u0430\u043a\u0438\u043c \u043e\u0431\u0440\u0430\u0437\u043e\u043c, \u043f\u043e\u043b\u0443\u0447\u0438\u0442\u0441\u044f \u043e\u0431\u044a\u0435\u043a\u0442 <code>DataStreamWriter<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u043e\u0441\u0442\u0430\u043b\u043e\u0441\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u0437\u0430\u043f\u0443\u0441\u0442\u0438\u0442\u044c:<\/p>\n<pre><code class=\"python\">console.start().awaitTermination()<\/code><\/pre>\n<p>\u041c\u0435\u0442\u043e\u0434 <code>start()<\/code> \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u0442 \u043f\u043e\u0442\u043e\u043a \u043d\u0430 \u0432\u044b\u043f\u043e\u043b\u043d\u0435\u043d\u0438\u0435. \u041d\u043e \u0434\u043b\u044f \u0442\u043e\u0433\u043e, \u0447\u0442\u043e\u0431\u044b \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0430 \u043d\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u0430\u0441\u044c, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u043f\u043e\u0441\u0442\u0430\u0432\u0438\u0442\u044c \u0435\u0451 \u0432 \u043e\u0436\u0438\u0434\u0430\u043d\u0438\u0435. \u042d\u0442\u043e \u043c\u043e\u0436\u043d\u043e \u0441\u0434\u0435\u043b\u0430\u0442\u044c \u043c\u0435\u0442\u043e\u0434\u043e\u043c <code>awaitTermination()<\/code>, \u043a\u043e\u0442\u043e\u0440\u044b\u0439 \u0431\u0443\u0434\u0435\u0442 \u043e\u0436\u0438\u0434\u0430\u0442\u044c, \u043f\u043e\u043a\u0430 \u043f\u043e\u0442\u043e\u043a \u043d\u0435 \u0431\u0443\u0434\u0435\u0442 \u0442\u0435\u0440\u043c\u0438\u043d\u0438\u0440\u043e\u0432\u0430\u043d.<\/p>\n<p>\u041f\u043e\u0441\u043c\u043e\u0442\u0440\u0438\u043c \u043d\u0430 \u0432\u044b\u0432\u043e\u0434, \u0437\u0430\u043f\u0438\u0441\u0430\u0432 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 &#171;<strong>Hello, World!<\/strong>&#187; \u0432 \u0442\u043e\u043f\u0438\u043a &#171;<strong>input00<\/strong>&#171;:<\/p>\n<pre><code>------------------------------------------- Batch: 1 ------------------------------------------- +-------------+------+ |\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 value|offset| +-------------+------+ |Hello, World!|\u00a0\u00a0\u00a0\u00a0 5| +-------------+------+<\/code><\/pre>\n<p>\u0413\u043e\u0442\u043e\u0432\u043e! \u0422\u0435\u043f\u0435\u0440\u044c \u043c\u044b \u043c\u043e\u0436\u0435\u043c \u0447\u0438\u0442\u0430\u0442\u044c \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f \u0438\u0437 Kafka.<\/p>\n<details class=\"spoiler\">\n<summary>\u041f\u043e\u043b\u043d\u044b\u0439 \u043a\u043e\u0434 \u043f\u0440\u0438\u043c\u0435\u0440\u0430<\/summary>\n<div class=\"spoiler__content\">\n<pre><code class=\"python\">def main(): \u00a0\u00a0\u00a0 from pyspark.sql import SparkSession  \u00a0\u00a0\u00a0 spark = (SparkSession \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .builder \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .appName('streaming-kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .getOrCreate()) \u00a0\u00a0\u00a0 spark.sparkContext.setLogLevel('WARN')  \u00a0\u00a0\u00a0 source = (spark \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .readStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('subscribe', 'input00,input01') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .load()) \u00a0\u00a0\u00a0 source.printSchema()  \u00a0\u00a0\u00a0 df = (source \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .selectExpr('CAST(value AS STRING)', 'offset'))  \u00a0\u00a0\u00a0 console = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('console') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 \u00a0\u00a0\u00a0\u00a0\u00a0.queryName('console output')) \u00a0\u00a0\u00a0 console.start().awaitTermination()   if __name__ == '__main__': \u00a0\u00a0\u00a0 main()<\/code><\/pre>\n<\/p>\n<\/div>\n<\/details>\n<h2>3) \u0417\u0430\u043f\u0443\u0441\u043a \u0441\u043a\u0440\u0438\u043f\u0442\u0430<\/h2>\n<p>\u0414\u043b\u044f \u0437\u0430\u043f\u0443\u0441\u043a\u0430 Spark \u043f\u0440\u0438\u043b\u043e\u0436\u0435\u043d\u0438\u0439 \u0438\u0441\u043f\u043e\u043b\u044c\u0437\u0443\u0435\u0442\u0441\u044f <strong>spark-submit<\/strong>. \u0427\u0442\u043e\u0431\u044b \u0432\u0437\u0430\u0438\u043c\u043e\u0434\u0435\u0439\u0441\u0442\u0432\u043e\u0432\u0430\u0442\u044c \u0441 Kafka, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0434\u043e\u0431\u0430\u0432\u0438\u0442\u044c \u0437\u0430\u0432\u0438\u0441\u0438\u043c\u043e\u0441\u0442\u044c &#171;<strong>spark-sql-kafka-0-10_2.12<\/strong>&#187; (\u0430\u043a\u0442\u0443\u0430\u043b\u044c\u043d\u0430\u044f \u0432\u0435\u0440\u0441\u0438\u044f \u0434\u043b\u044f Spark 3.2.1 \u043d\u0430 \u043c\u043e\u043c\u0435\u043d\u0442 \u043d\u0430\u043f\u0438\u0441\u0430\u043d\u0438\u044f \u0441\u0442\u0430\u0442\u044c\u0438) \u043a \u0430\u0440\u0433\u0443\u043c\u0435\u043d\u0442\u0443 &#171;<strong>&#8212;packages<\/strong>&#187; (\u0435\u0441\u043b\u0438 \u0435\u0441\u0442\u044c \u0434\u043e\u0441\u0442\u0443\u043f \u043a \u0441\u0435\u0442\u0438 \u0418\u043d\u0442\u0435\u0440\u043d\u0435\u0442 \u0438\u043b\u0438 \u043f\u0430\u043a\u0435\u0442 \u0443\u0436\u0435 \u0438\u043c\u0435\u0435\u0442\u0441\u044f \u0432 \u043a\u044d\u0448\u0435) \u0438\u043b\u0438 &#171;<strong>&#8212;jars<\/strong>&#187; (\u0435\u0441\u043b\u0438 \u0438\u043c\u0435\u0435\u0442\u0441\u044f \u043b\u043e\u043a\u0430\u043b\u044c\u043d\u043e \u043d\u0430 \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0435\u043c\u043e\u0439 \u043c\u0430\u0448\u0438\u043d\u0435, \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u043f\u0443\u0442\u044c).<\/p>\n<p><strong>spark-submit<\/strong> \u043f\u0440\u0438\u043d\u0438\u043c\u0430\u0435\u0442 \u043c\u043d\u043e\u0436\u0435\u0441\u0442\u0432\u043e \u043f\u0430\u0440\u0430\u043c\u0435\u0442\u0440\u043e\u0432, \u043e \u043a\u043e\u0442\u043e\u0440\u044b\u0445 \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0432 <a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/submitting-applications.html\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>. \u042d\u0442\u0430 \u0443\u0442\u0438\u043b\u0438\u0442\u0430 \u0434\u043e\u0441\u0442\u0443\u043f\u043d\u0430 \u0432 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u0441\u0440\u0435\u0434\u0435 Python, \u0433\u0434\u0435 \u0431\u044b\u043b \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d \u043f\u0430\u043a\u0435\u0442 PySpark, \u043f\u043e\u044d\u0442\u043e\u043c\u0443 \u0435\u0451 \u043c\u043e\u0436\u043d\u043e \u0437\u0430\u043f\u0443\u0441\u043a\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u043e\u0439 \u043a\u043e\u043c\u0430\u043d\u0434\u043e\u0439 \u0432 \u0442\u0435\u0440\u043c\u0438\u043d\u0430\u043b\u0435 \u043f\u0440\u0438 \u0430\u043a\u0442\u0438\u0432\u0438\u0440\u043e\u0432\u0430\u043d\u043d\u043e\u0439 \u0432\u0438\u0440\u0442\u0443\u0430\u043b\u044c\u043d\u043e\u0439 \u0441\u0440\u0435\u0434\u0435 (\u0434\u043e\u043b\u0436\u043d\u0430 \u0431\u044b\u0442\u044c \u0443\u0441\u0442\u0430\u043d\u043e\u0432\u043b\u0435\u043d\u0430 Java):<\/p>\n<pre><code class=\"bash\">spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 example1.py<\/code><\/pre>\n<h2>4) \u041f\u0438\u0448\u0435\u043c \u0432 \u0442\u043e\u043f\u0438\u043a Kafka<\/h2>\n<p>\u0412\u044b\u0432\u043e\u0434 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka \u043d\u0430\u043f\u0438\u0441\u0430\u0442\u044c \u043d\u0435 \u043c\u043d\u043e\u0433\u0438\u043c \u0441\u043b\u043e\u0436\u043d\u0435\u0435, \u0447\u0435\u043c \u0432\u044b\u0432\u043e\u0434 \u0432 \u043a\u043e\u043d\u0441\u043e\u043b\u044c.<\/p>\n<p>\u0414\u043e\u043f\u043e\u043b\u043d\u044f\u044f \u043f\u0440\u0435\u0434\u044b\u0434\u0443\u0449\u0438\u0439 \u043f\u0440\u0438\u043c\u0435\u0440, \u043d\u0430\u043f\u0438\u0448\u0435\u043c \u0435\u0449\u0451 \u043e\u0434\u0438\u043d \u0432\u044b\u0445\u043e\u0434 \u0434\u043b\u044f <code>df<\/code>:<\/p>\n<pre><code class=\"python\">query = (df \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .writeStream \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .format('kafka') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .queryName('kafka-output') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('kafka.bootstrap.servers', 'localhost:9092') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('topic', 'output00') \u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .option('checkpointLocation', '.\/.local\/checkpoint'))<\/code><\/pre>\n<p>\u0414\u043b\u044f \u0432\u044b\u0432\u043e\u0434\u0430 \u0432 Kafka \u043d\u0435\u043e\u0431\u0445\u043e\u0434\u0438\u043c\u043e \u0432 \u043c\u0435\u0442\u043e\u0434 <code>format()<\/code> \u0437\u0430\u043f\u0438\u0441\u0430\u0442\u044c &#171;<strong>kafka<\/strong>&#187; \u0438 \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0432 \u043e\u043f\u0446\u0438\u044f\u0445:<\/p>\n<ul>\n<li>\n<p><strong>kafka.bootstrap.servers <\/strong>\u2013 \u0430\u0434\u0440\u0435\u0441 \u0431\u0440\u043e\u043a\u0435\u0440 \u0441\u0435\u0440\u0432\u0435\u0440\u0430 (\u043c\u043e\u0436\u043d\u043e \u043d\u0435\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u2013 \u00a0\u0447\u0435\u0440\u0435\u0437 \u0437\u0430\u043f\u044f\u0442\u0443\u044e \u0432\u043d\u0443\u0442\u0440\u0438 \u0441\u0442\u0440\u043e\u043a\u0438);<\/p>\n<\/li>\n<li>\n<p><strong>topic<\/strong> \u2013 \u00a0\u0432 \u043a\u0430\u043a\u043e\u0439 \u0442\u043e\u043f\u0438\u043a \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c \u0434\u0430\u043d\u043d\u044b\u0435 (\u0442\u043e\u043b\u044c\u043a\u043e \u043e\u0434\u0438\u043d);<\/p>\n<\/li>\n<li>\n<p><strong>checkpointLocation<\/strong> \u2013 \u0443\u043a\u0430\u0437\u0430\u0442\u044c \u0434\u0438\u0440\u0435\u043a\u0442\u043e\u0440\u0438\u044e, \u0433\u0434\u0435 \u0431\u0443\u0434\u0443\u0442 \u0437\u0430\u043f\u0438\u0441\u044b\u0432\u0430\u0442\u044c\u0441\u044f\/\u0447\u0438\u0442\u0430\u0442\u044c\u0441\u044f \u0447\u0435\u043a\u043f\u043e\u0438\u043d\u0442\u044b. \u0427\u0435\u043a\u043f\u043e\u0438\u043d\u0442\u044b \u0441\u043e\u0445\u0440\u0430\u043d\u044f\u044e\u0442 \u0443\u043a\u0430\u0437\u0430\u0442\u0435\u043b\u044c \u043d\u0430 \u043c\u0435\u0441\u0442\u0435, \u0433\u0434\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u0430\u0441\u044c \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0430 \u0438\u0441\u0442\u043e\u0447\u043d\u0438\u043a\u0430, \u0447\u0442\u043e \u043c\u043e\u0436\u0435\u0442 \u043f\u0440\u0438\u0433\u043e\u0434\u0438\u0442\u044c\u0441\u044f \u043f\u0440\u0438 \u043f\u0435\u0440\u0435\u0437\u0430\u043f\u0443\u0441\u043a\u0435 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u044b (\u043f\u0440\u043e\u0434\u043e\u043b\u0436\u0438\u0442 \u0441 \u0442\u043e\u0433\u043e \u043c\u0435\u0441\u0442\u0430, \u0433\u0434\u0435 \u043e\u0441\u0442\u0430\u043d\u043e\u0432\u0438\u043b\u0441\u044f). \u041d\u0430\u0441\u0442\u0440\u0430\u0438\u0432\u0430\u0435\u0442\u0441\u044f \u043d\u0430 <code>DataStreamWriter<\/code>. \u0411\u043e\u043b\u0435\u0435 \u043f\u043e\u0434\u0440\u043e\u0431\u043d\u043e \u043f\u0440\u043e \u0447\u0435\u043a\u043f\u043e\u0438\u043d\u0442\u044b \u043c\u043e\u0436\u043d\u043e \u043f\u0440\u043e\u0447\u0438\u0442\u0430\u0442\u044c \u0432 <a href=\"https:\/\/spark.apache.org\/docs\/3.2.1\/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing\">\u0434\u043e\u043a\u0443\u043c\u0435\u043d\u0442\u0430\u0446\u0438\u0438<\/a>.<\/p>\n<\/li>\n<\/ul>\n<p>\u0415\u0449\u0451 \u0434\u043e\u0431\u0430\u0432\u043e\u0447\u043d\u043e \u0437\u0434\u0435\u0441\u044c \u043f\u043e\u044f\u0432\u0438\u043b\u0430\u0441\u044c \u0441\u0442\u0440\u043e\u043a\u0430 \u0441 \u043c\u0435\u0442\u043e\u0434\u043e\u043c <code>queryName()<\/code>. \u042d\u0442\u043e \u043d\u0435\u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e, \u043d\u043e \u0442\u0430\u043a \u043a\u0430\u043a \u0443 \u043d\u0430\u0441 \u043f\u043e\u044f\u0432\u0438\u043b\u0441\u044f \u0432\u0442\u043e\u0440\u043e\u0439 \u043f\u043e\u0442\u043e\u043a \u0432 \u043e\u0434\u043d\u043e\u0439 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0435, \u0441\u043b\u0435\u0434\u0443\u0435\u0442 \u0438\u0445 \u0432\u0441\u0435 \u043f\u0440\u043e\u0438\u043c\u0435\u043d\u043e\u0432\u0430\u0442\u044c, \u0447\u0442\u043e\u0431\u044b \u0440\u0430\u0437\u043b\u0438\u0447\u0430\u0442\u044c \u043f\u0440\u0438 \u043c\u043e\u043d\u0438\u0442\u043e\u0440\u0438\u043d\u0433\u0435. \u0417\u0430\u043f\u0443\u0441\u0442\u0438\u043c \u043e\u0431\u0430 \u043f\u043e\u0442\u043e\u043a\u0430:<\/p>\n<pre><code class=\"python\">console.start() query.start().awaitTermination()<\/code><\/pre>\n<p>\u0412\u044b\u0432\u043e\u0434:<\/p>\n<pre><code>------------------------------------------- Batch: 1 ------------------------------------------- +-------------------+------+ |\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 value|offset| +-------------------+------+ |Hello, second sink!|\u00a0\u00a0\u00a0\u00a0 0| +-------------------+------+<\/code><\/pre>\n<p>\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0442\u043e\u043f\u0438\u043a Kafka \u0442\u0430\u043a\u0436\u0435 \u0437\u0430\u043f\u0438\u0441\u0430\u043d:<\/p>\n<figure class=\"full-width\"><figcaption>\u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 \u0432 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u043c \u0442\u043e\u043f\u0438\u043a\u0435 Kafka<\/figcaption><\/figure>\n<p><strong>\u0412\u0410\u0416\u041d\u041e<\/strong>: \u0443 \u0432\u044b\u0445\u043e\u0434\u043d\u043e\u0433\u043e <code>DataFrame<\/code> \u043e\u0431\u044f\u0437\u0430\u0442\u0435\u043b\u044c\u043d\u043e \u0434\u043e\u043b\u0436\u043d\u043e \u0431\u044b\u0442\u044c \u043f\u043e\u043b\u0435 &#171;<strong>value<\/strong>&#171;, \u0438\u043d\u0430\u0447\u0435 \u043f\u0440\u043e\u0433\u0440\u0430\u043c\u043c\u0430 \u0443\u043f\u0430\u0434\u0451\u0442 \u0441 \u043e\u0448\u0438\u0431\u043a\u043e\u0439. \u0418\u043c\u0435\u043d\u043d\u043e \u044d\u0442\u043e \u043f\u043e\u043b\u0435 \u0438 \u0431\u0443\u0434\u0435\u0442 \u0437\u0430\u043f\u0438\u0441\u0430\u043d\u043e \u0432 &#171;<strong>value<\/strong>&#171;<\/p>\n<\/div>\n<\/div>\n<\/div>\n<\/div>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[],"tags":[],"class_list":["post-335296","post","type-post","status-publish","format-standard","hentry"],"_links":{"self":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/335296","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=335296"}],"version-history":[{"count":0,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=\/wp\/v2\/posts\/335296\/revisions"}],"wp:attachment":[{"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=335296"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=335296"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/savepearlharbor.com\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=335296"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}