{"id":6116,"date":"2023-05-08T15:02:27","date_gmt":"2023-05-08T15:02:27","guid":{"rendered":"https:\/\/bentego.com\/apache-spark-structured-streaming-via-docker-compose\/"},"modified":"2025-07-28T10:53:38","modified_gmt":"2025-07-28T10:53:38","slug":"apache-spark-structured-streaming-via-docker-compose","status":"publish","type":"post","link":"https:\/\/bentego.com\/tr\/apache-spark-structured-streaming-via-docker-compose\/","title":{"rendered":"Docker Compose ile Apache Spark Yap\u0131land\u0131r\u0131lm\u0131\u015f Ak\u0131\u015f"},"content":{"rendered":"

Bir veri hatt\u0131 olu\u015fturmak, \u00f6zellikle ta\u015f\u0131nabilirlik, esneklik, \u00f6l\u00e7eklenebilirlik vb. hususlar\u0131 g\u00f6z \u00f6n\u00fcnde bulundurman\u0131z gerekti\u011finde zorlay\u0131c\u0131 olabilir. Bu zorluklar\u0131n \u00fcstesinden gelmek i\u00e7in docker iyi bilinen bir \u00e7\u00f6z\u00fcmd\u00fcr. Bu makalede, docker-compose dosyas\u0131 arac\u0131l\u0131\u011f\u0131yla bir veri hatt\u0131 olu\u015fturma hakk\u0131nda konu\u015faca\u011f\u0131z. <\/p>\n

\"\"<\/a><\/p>\n

0. Installation Processes<\/strong><\/pre>\n
Bu projeyi ger\u00e7ekle\u015ftirmek i\u00e7in gerekli t\u00fcm bile\u015fenleri verilen ad\u0131mlar\u0131 kullanarak y\u00fckleyebilirsiniz.<\/p>\n
Installation of ROS<\/strong><\/pre>\n
ROS’un t\u00fcm kurulum s\u00fcrecine de\u011finmeyece\u011fiz, ancak gerekli t\u00fcm bilgilere \u015fu adresten eri\u015febilirsiniz ROS Noetic & Ubuntu 20.04 Kurulumu<\/a>.<\/span><\/span><\/p>\n
Installation of Docker on Ubuntu<\/strong><\/pre>\n
Bunu kullanabilirsin URL<\/a><\/span><\/p>\n
Installation of Kafka-Python Library used for publishing data received from ROS to Kafka<\/strong><\/pre>\n
\u2757 E\u011fer y\u00fcklemediyseniz <\/span><\/span><\/em>kafka-python<\/em><\/a><\/span><\/span>verilen komutu kullan\u0131n ve ard\u0131ndan verilen dosyalar\u0131 \u00e7al\u0131\u015ft\u0131r\u0131n.<\/em><\/p>\n
\n
pip install kafka-python\n<\/span><\/pre>\n<\/blockquote>\n
1. Prepare a robotic simulation environment<\/strong><\/pre>\n
ROS (Robot \u0130\u015fletim Sistemi)<\/a><\/span><\/span>robotik bir ortam tasarlamam\u0131z\u0131 sa\u011flar. Bu projede veri sa\u011flay\u0131c\u0131 olarak ROS’u kullanaca\u011f\u0131z. “odom” bir arac\u0131n konumunu temsil eden bir mesaj t\u00fcr\u00fcd\u00fcr. Rastgele “odom” verileri \u00fcreten ve bunlar\u0131 yay\u0131nlayan verilen kodu kullan\u0131yoruz.   <\/p>\n
\n
#!\/usr\/bin\/env python3\nimport math\nfrom math import sin, cos, pi\nimport rospy\nimport tf\nfrom nav_msgs.msg import Odometry\nfrom geometry_msgs.msg import Point, Pose, Quaternion, Twist, Vector3\nrospy.init_node('odometry_publisher')\nodom_pub = rospy.Publisher(\"odom\", Odometry, queue_size=50)\nodom_broadcaster = tf.TransformBroadcaster()\nx = 0.0\ny = 0.0\nth = 0.0\nvx = 0.1\nvy = -0.1\nvth = 0.1\ncurrent_time = rospy.Time.now()\nlast_time = rospy.Time.now()\nr = rospy.Rate(1.0)\nwhile not rospy.is_shutdown():\n    current_time = rospy.Time.now()\n    # compute odometry in a typical way given the velocities of the robot\n    dt = (current_time - last_time).to_sec()\n    delta_x = (vx * cos(th) - vy * sin(th)) * dt\n    delta_y = (vx * sin(th) + vy * cos(th)) * dt\n    delta_th = vth * dt\n    x += delta_x\n    y += delta_y\n    th += delta_th\n    # since all odometry is 6DOF we'll need a quaternion created from yaw\n    odom_quat = tf.transformations.quaternion_from_euler(0, 0, th)\n    # first, we'll publish the transform over tf\n    odom_broadcaster.sendTransform(\n        (x, y, 0.),\n        odom_quat,\n        current_time,\n        \"base_link\",\n        \"odom\"\n    )\n    # next, we'll publish the odometry message over ROS\n    odom = Odometry()\n    odom.header.stamp = current_time\n    odom.header.frame_id = \"odom\"\n    # set the position\n    odom.pose.pose = Pose(Point(x, y, 0.), Quaternion(*odom_quat))\n    # set the velocity\n    odom.child_frame_id = \"base_link\"\n    odom.twist.twist = Twist(Vector3(vx, vy, 0), Vector3(0, 0, vth))\n    # publish the message\n    odom_pub.publish(odom)\n    last_time = current_time\n    r.sleep()\n\n<\/pre>\n<\/blockquote>\n
Run the given code and analysis the data we will use<\/strong><\/pre>\n
Bu betik odometri verilerini ROS “odom” ba\u015fl\u0131\u011f\u0131 ile yay\u0131nlar. B\u00f6ylece, yay\u0131nlanan verileri verilen komutla g\u00f6rebiliriz: <\/p>\n
\n
# run the script environment\npython3 odomPublisher.py<\/span># check the topic to see data\nrostopic echo \/odom<\/span><\/pre>\n<\/blockquote>\n
Bu kullan\u0131m durumunda, verilerin sadece verilen k\u0131sm\u0131yla ilgilenece\u011fiz:<\/p>\n
\n
position: \n      x: -2.000055643960576\n      y: -0.4997879642933192\n      z: -0.0010013932644100873\n    orientation: \n      x: -1.3486164084605e-05\n      y: 0.0038530870521455017\n      z: 0.0016676819550213058\n      w: 0.9999911861487526<\/span><\/pre>\n<\/blockquote>\n
2. Prepare Docker-Compose File<\/strong><\/pre>\n
\u00d6ncelikle mimari i\u00e7in datapipeline ad\u0131nda bir a\u011f olu\u015fturduk. Mimari 4 servisten olu\u015fuyor ve her birinin statik bir IP adresi var ve a\u015fa\u011f\u0131da verildi\u011fi gibi varsay\u0131lan portu kullan\u0131yor: <\/p>\n
K\u0131v\u0131lc\u0131m: 172.18.0.2
\nZookeeper: 172.18.0.3
\nKafka: 172.18.0.4
\nCassandra: 172.18.0.5<\/p><\/blockquote>\n
Komut dosyalar\u0131m\u0131z\u0131 konteynerlere aktarmak i\u00e7in “volumes” kullan\u0131yoruz.<\/p>\n
\u2757 Sisteminiz i\u00e7in ” ..\/streamingProje:\/home” k\u0131sm\u0131n\u0131 uygulaman\u0131z gerekir.<\/em><\/p>\n
\u015eu adrese eri\u015febilirsiniz docker-compose<\/span><\/a>  <\/span>ve yap\u0131land\u0131rmalar\u0131n\u0131z\u0131 de\u011fi\u015ftirin.<\/p>\n
\n
\n
\n
\n
version: '3'\nnetworks:\n    datapipeline:\n        driver: bridge\n        ipam:\n            driver: default\n            config:\n                - subnet: \"172.18.0.0\/16\"\n\nservices:  \n  spark:\n    image: docker.io\/bitnami\/spark:3\n    container_name: spark_master\n    hostname: spark_master\n    user: root\n    environment:\n      - SPARK_MODE=master\n      - SPARK_RPC_AUTHENTICATION_ENABLED=no\n      - SPARK_RPC_ENCRYPTION_ENABLED=no\n      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no\n      - SPARK_SSL_ENABLED=no\n    ports:\n      - '8080:8080'\n    volumes:\n      - ..\/streamingProje:\/home\n      - \/opt\/spark\/conf\/spark-defaults.conf:\/opt\/bitnami\/spark\/conf\/spark-defaults.conf\n      - \/opt\/spark\/jars:\/opt\/bitnami\/spark\/ivy:z\n    networks:\n      datapipeline:\n        ipv4_address: 172.18.0.2\n  zookeeper:\n    image: 'bitnami\/zookeeper:latest'\n    container_name: zookeeper\n    hostname: zookeeper\n    ports:\n      - '2181:2181'\n    environment:\n      - ALLOW_ANONYMOUS_LOGIN=yes\n    networks:\n      datapipeline:\n        ipv4_address: 172.18.0.3\n  kafka:\n    image: 'bitnami\/kafka:latest'\n    container_name: kafka\n    hostname: kafka\n    ports:\n      - '9092:9092'\n    environment:\n      - KAFKA_BROKER_ID=1\n      - KAFKA_CFG_LISTENERS=PLAINTEXT:\/\/172.18.0.4:9092\n      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT:\/\/172.18.0.4:9092\n      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181\n      - ALLOW_PLAINTEXT_LISTENER=yes\n    depends_on:\n      - zookeeper\n    volumes:\n      - ..\/streamingProje:\/home\n    networks:\n      datapipeline:\n        ipv4_address: 172.18.0.4\n  cassandra:\n    image: 'bitnami\/cassandra:latest'\n    container_name: cassandra\n    hostname: cassandra\n    ports:\n      - '9042:9042'\n    volumes:\n      - ..\/streamingProje:\/home\n    networks:\n      datapipeline:\n        ipv4_address: 172.18.0.5<\/pre>\n<\/blockquote>\n<\/div>\n<\/div>\n<\/figure>\n
3. Running docker-compose file<\/strong><\/pre>\n
Sa\u011flanan t\u00fcm dosyalar\u0131 i\u00e7eren \u00e7al\u0131\u015fma alan\u0131 klas\u00f6r\u00fcn\u00fcz\u00fc a\u00e7\u0131n ve verilen komutu a\u015fa\u011f\u0131daki gibi \u00e7al\u0131\u015ft\u0131r\u0131n.<\/p>\n
\n
# run docker-compose file\ndocker-compose up<\/span><\/pre>\n<\/blockquote>\n
\u015e\u00f6yle bir manzaraya sahip olacaks\u0131n\u0131z:<\/p>\n
\n
\n
<\/source><\/source><\/picture>\"\"<\/a><\/div>\n<\/div>\n<\/figure>\n
Sonu\u00e7ta, konteyner \u00e7al\u0131\u015f\u0131yor, ortam\u0131n\u0131z\u0131 kurabilirsiniz.<\/p>\n
Prepare Kafka for Use Case<\/strong><\/pre>\n
\u00d6ncelikle verilen komutlar\u0131 kullanarak ROS odom verileri i\u00e7in odometry<\/em> ad\u0131nda yeni bir Kafka konusu olu\u015fturaca\u011f\u0131z:<\/p>\n
\n
# Execute kafka container with container id given above\ndocker exec -it 1c31511ce206 bash<\/span># Create Kafka \"odometry\" topic for ROS odom data\nkafka$ bin\/kafka-topics.sh --create --topic odom --partitions 1 --replication-factor 1 \n-bootstrap-server localhost:9092<\/span><\/pre>\n<\/blockquote>\n
Check Kafka setup through Zookeeper<\/strong><\/pre>\n
\n
# Execute zookeeper container with container id given above\ndocker exec -it 1c31511ce206 bash<\/span># run command\nopt\/bitnami\/zookeeper\/bin\/zkCli.sh -server localhost:2181 \n<\/span># list all brokers topic\nls \/brokers\/topics<\/span><\/pre>\n<\/blockquote>\n
\u015e\u00f6yle bir manzaraya sahip olacaks\u0131n\u0131z:<\/p>\n
\n
<\/source><\/source><\/picture>\"\"<\/a><\/div>\n<\/figure>\n
Prepare Cassandra for Use Case<\/strong><\/pre>\n
\u0130lk olarak, verilen komutu kullanarak bir anahtar alan\u0131<\/em> ve ard\u0131ndan i\u00e7inde bir konu<\/em> olu\u015fturaca\u011f\u0131z:<\/p>\n
\n
# Execute cassandra container with container id given above\ndocker exec -it 1c31511ce206 bash\n\n<\/span># Open the cqlsh\ncqlsh -u cassandra -p cassandra<\/span># Run the command to create 'ros' keyspace\n\ncqlsh> CREATE KEYSPACE ros WITH replication = {\n'class':'SimpleStrategy', \n'replication_factor' : 1};\n\n<\/span># Then, run the command to create 'odometry' topic in 'ros'\ncqlsh> create table ros.odometry(\n        id int primary key, \n        posex float,\n        posey float,\n        posez float,\n        orientx float,\n        orienty float,\n        orientz float,\n        orientw float);<\/span># Check your setup is correct\ncqlsh> DESCRIBE ros.odometry<\/span><\/pre>\n<\/blockquote>\n
\u26a0\ufe0f Konunun i\u00e7eri\u011fi Spark \u015femas\u0131 ile ayn\u0131 olmal\u0131d\u0131r: Burada \u00e7ok dikkatli olun!<\/em><\/p>\n
4. Prepare Apache Spark structured streaming<\/strong><\/pre>\n
Analiz sonu\u00e7lar\u0131n\u0131 konsola ya da Cassandra’ya yazabilirsiniz.<\/p>\n
(First Way) Prepare Apache Spark Structured Streaming Pipeline Kafka to Cassandra<\/strong><\/pre>\n
Odometri<\/em> konular\u0131n\u0131 Kafka’dan okuyan, analiz eden ve ard\u0131ndan sonu\u00e7lar\u0131 Cassandra’ya yazan bir ak\u0131\u015f beti\u011fi yazaca\u011f\u0131z. Kullanaca\u011f\u0131m\u0131z streamingKafka2Cassandra.py<\/a><\/span> <\/span>bunu yapmak i\u00e7in. <\/p>\n
\u00d6ncelikle, Cassandra’da daha \u00f6nce tan\u0131mlad\u0131\u011f\u0131m\u0131z gibi bir \u015fema olu\u015fturuyoruz.<\/p>\n
\u26a0\ufe0f \u015eema i\u00e7eri\u011fi Casssandra tablosu ile ayn\u0131 olmal\u0131d\u0131r: Burada \u00e7ok dikkatli olun!<\/em><\/p>\n
\n
odometrySchema = StructType([\n                StructField(\"id\",IntegerType(),False),\n                StructField(\"posex\",FloatType(),False),\n                StructField(\"posey\",FloatType(),False),\n                StructField(\"posez\",FloatType(),False),\n                StructField(\"orientx\",FloatType(),False),\n                StructField(\"orienty\",FloatType(),False),\n                StructField(\"orientz\",FloatType(),False),\n                StructField(\"orientw\",FloatType(),False)\n            ])<\/span><\/pre>\n<\/blockquote>\n
Ard\u0131ndan, bir Spark Oturumu olu\u015fturuyoruz ve yap\u0131land\u0131rmam\u0131z\u0131 burada belirtiyoruz:<\/p>\n
\n
spark = SparkSession \\\n    .builder \\\n    .appName(\"SparkStructuredStreaming\") \\\n    .config(\"spark.cassandra.connection.host\",\"172.18.0.5\")\\\n    .config(\"spark.cassandra.connection.port\",\"9042\")\\\n    .config(\"spark.cassandra.auth.username\",\"cassandra\")\\\n    .config(\"spark.cassandra.auth.password\",\"cassandra\")\\\n    .config(\"spark.driver.host\", \"localhost\")\\\n    .getOrCreate()<\/span><\/pre>\n<\/blockquote>\n
Kafka ak\u0131\u015f\u0131n\u0131 okumak i\u00e7in readStream() kullan\u0131yoruz ve Kafka konfig\u00fcrasyonlar\u0131n\u0131 a\u015fa\u011f\u0131da verildi\u011fi gibi belirliyoruz:<\/p>\n
\n
df = spark \\\n  .readStream \\\n  .format(\"kafka\") \\\n  .option(\"kafka.bootstrap.servers\", \"172.18.0.4:9092\") \\\n  .option(\"subscribe\", \"rosmsgs\") \\\n  .option(\"delimeter\",\",\") \\\n  .option(\"startingOffsets\", \"earliest\") \\\n  .load()<\/span><\/pre>\n<\/blockquote>\n
Kafka verileri binary olarak g\u00f6nderdi\u011finden, \u00f6ncelikle a\u015fa\u011f\u0131da verildi\u011fi gibi selectExpr() kullanarak binary de\u011feri String’e d\u00f6n\u00fc\u015ft\u00fcrmemiz gerekir:<\/p>\n
\n
df1 = df.selectExpr(\"CAST(value AS STRING)\").select(from_json(col(\"value\"),\nodometrySchema).alias(\"data\")).select(\"data.*\")\ndf1.printSchema()<\/span><\/pre>\n<\/blockquote>\n
Apache Spark hen\u00fcz ak\u0131\u015f verilerini do\u011frudan Cassandra’ya yazma yetene\u011fine sahip olmasa da (writeStream() kullanarak), foreachBatch() kullanarak bunu a\u015fa\u011f\u0131da verildi\u011fi gibi yapabiliriz:<\/p>\n
\n
def writeToCassandra(writeDF, _):\n  writeDF.write \\\n    .format(\"org.apache.spark.sql.cassandra\")\\\n    .mode('append')\\\n    .options(table=\"odometry\", keyspace=\"ros\")\\\n    .save()\n\n<\/span>df1.writeStream \\\n    .foreachBatch(writeToCassandra) \\\n    .outputMode(\"update\") \\\n    .start()\\\n    .awaitTermination()\ndf1.show()<\/span><\/pre>\n<\/blockquote>\n
Son olarak, verilen komut dosyas\u0131n\u0131 elde ettik streamingKafka2Cassandra.py<\/a><\/span><\/span>:<\/em><\/strong><\/p>\n
\n
\n
\n
\n
from pyspark.sql import SparkSession\nfrom pyspark.sql.types import StructType,StructField,FloatType,IntegerType\nfrom pyspark.sql.functions import from_json,col\nodometrySchema = StructType([\n                StructField(\"id\",IntegerType(),False),\n                StructField(\"posex\",FloatType(),False),\n                StructField(\"posey\",FloatType(),False),\n                StructField(\"posez\",FloatType(),False),\n                StructField(\"orientx\",FloatType(),False),\n                StructField(\"orienty\",FloatType(),False),\n                StructField(\"orientz\",FloatType(),False),\n                StructField(\"orientw\",FloatType(),False)\n            ])\nspark = SparkSession \\\n    .builder \\\n    .appName(\"SparkStructuredStreaming\") \\\n    .config(\"spark.cassandra.connection.host\",\"172.18.0.5\")\\\n    .config(\"spark.cassandra.connection.port\",\"9042\")\\\n    .config(\"spark.cassandra.auth.username\",\"cassandra\")\\\n    .config(\"spark.cassandra.auth.password\",\"cassandra\")\\\n    .config(\"spark.driver.host\", \"localhost\")\\\n    .getOrCreate()\nspark.sparkContext.setLogLevel(\"ERROR\")\n\ndf = spark \\\n  .readStream \\\n  .format(\"kafka\") \\\n  .option(\"kafka.bootstrap.servers\", \"172.18.0.4:9092\") \\\n  .option(\"subscribe\", \"rosmsgs\") \\\n  .option(\"delimeter\",\",\") \\\n  .option(\"startingOffsets\", \"earliest\") \\\n  .load() \ndf.printSchema()\ndf1 = df.selectExpr(\"CAST(value AS STRING)\").select(from_json(col(\"value\"),\nodometrySchema).alias(\"data\")).select(\"data.*\")\ndf1.printSchema()\n\ndef writeToCassandra(writeDF, _):\n  writeDF.write \\\n    .format(\"org.apache.spark.sql.cassandra\")\\\n    .mode('append')\\\n    .options(table=\"odometry\", keyspace=\"ros\")\\\n    .save()\ndf1.writeStream \\\n    .foreachBatch(writeToCassandra) \\\n    .outputMode(\"update\") \\\n    .start()\\\n    .awaitTermination()<\/pre>\n<\/blockquote>\n<\/div>\n<\/div>\n<\/figure>\n
(Second Way) Prepare Apache Spark Structured Streaming Pipeline Kafka to Console<\/strong><\/pre>\n
Konsola yazmak ile Cassandra’ya yazmak aras\u0131nda birka\u00e7 fark vard\u0131r. Ak\u0131\u015f\u0131 do\u011frudan konsola yaz\u0131yoruz. writeStream() ile ak\u0131\u015f verilerini do\u011frudan konsola yazabiliriz.  <\/p>\n
\n
df1.writeStream \\\n  .outputMode(\"update\") \\\n  .format(\"console\") \\\n  .option(\"truncate\", False) \\\n  .start() \\\n  .awaitTermination()<\/span><\/pre>\n<\/blockquote>\n
S\u00fcrecin geri kalan\u0131 bir \u00f6ncekiyle ayn\u0131 \u015fekilde ger\u00e7ekle\u015fir. Son olarak, verilen komut dosyas\u0131n\u0131 elde ettik streamingKafka2Console.py<\/a><\/span><\/span>:<\/strong><\/em> <\/p>\n
\n
\n
\n
\n
from pyspark.sql import SparkSession\nfrom pyspark.sql.types import StructType,StructField,IntegerType,FloatType\nfrom pyspark.sql.functions import split,from_json,col\nodometrySchema = StructType([\n                StructField(\"id\",IntegerType(),False),\n                StructField(\"posex\",FloatType(),False),\n                StructField(\"posey\",FloatType(),False),\n                StructField(\"posez\",FloatType(),False),\n                StructField(\"orientx\",FloatType(),False),\n                StructField(\"orienty\",FloatType(),False),\n                StructField(\"orientz\",FloatType(),False),\n                StructField(\"orientw\",FloatType(),False)\n            ])\nspark = SparkSession \\\n    .builder \\\n    .appName(\"SSKafka\") \\\n    .config(\"spark.driver.host\", \"localhost\")\\\n    .getOrCreate()\nspark.sparkContext.setLogLevel(\"ERROR\")\ndf = spark \\\n  .readStream \\\n  .format(\"kafka\") \\\n  .option(\"kafka.bootstrap.servers\", \"172.18.0.4:9092\") \\\n  .option(\"subscribe\", \"rosmsgs\") \\\n  .option(\"delimeter\",\",\") \\\n  .option(\"startingOffsets\", \"earliest\") \\\n  .load() \ndf.printSchema()\ndf1 = df.selectExpr(\"CAST(value AS STRING)\").select(from_json(col(\"value\"),\nodometrySchema).alias(\"data\")).select(\"data.*\")\ndf1.printSchema()\ndf1.writeStream \\\n  .outputMode(\"update\") \\\n  .format(\"console\") \\\n  .option(\"truncate\", False) \\\n  .start() \\\n  .awaitTermination()<\/pre>\n<\/blockquote>\n<\/div>\n<\/div>\n<\/figure>\n
5. Demonstration & Results<\/strong><\/pre>\n
T\u00fcm haz\u0131rl\u0131klar\u0131n tamamland\u0131\u011f\u0131ndan eminseniz, bir demo ba\u015flatabilirsiniz. Verilen ad\u0131mlar\u0131 takip etmelisiniz. <\/p>\n
Start ROS and publish odom data to Kafka.<\/strong><\/pre>\n