{"id":5147,"date":"2023-05-08T15:02:27","date_gmt":"2023-05-08T12:02:27","guid":{"rendered":"https:\/\/bentego.com\/?p=5147"},"modified":"2025-10-20T16:09:19","modified_gmt":"2025-10-20T16:09:19","slug":"apache-spark-structured-streaming-via-docker-compose","status":"publish","type":"post","link":"https:\/\/bentego.com\/en\/apache-spark-structured-streaming-via-docker-compose\/","title":{"rendered":"Apache Spark Structured Streaming Via Docker Compose"},"content":{"rendered":"

Building a data pipeline could be challenging especially when you have to take into account portability, flexibility, scalability etc. To overcome these challenges docker is one the well-known solution. In this article, we are gonna talk about building a data pipeline via docker-compose file.<\/p>\n

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

0. Installation Processes<\/strong><\/pre>\n
You are able to install all required components to realize this project using the given steps.<\/p>\n
Installation of ROS<\/strong><\/pre>\n
We won\u2019t address the whole installation process of ROS but you can access all required info from\u00a0ROS Noetic & Ubuntu 20.04 Installation<\/a>.<\/span><\/span><\/p>\n
Installation of Docker on Ubuntu<\/strong><\/pre>\n
You can utilize this\u00a0URL<\/a><\/span><\/p>\n
Installation of Kafka-Python Library used for publishing data received from ROS to Kafka<\/strong><\/pre>\n
\u2757 If you haven\u2019t installed\u00a0<\/span><\/span><\/em>kafka-python<\/em><\/a><\/span><\/span>,<\/strong> use the given command and then run given files.<\/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 Operating System)<\/a><\/span><\/span>allows us to design a robotic environment. In this project, we will use ROS as a data provider. \u201codom\u201d is a type of message that represents the position of a vehicle. We utilize the given code that generates arbitrary \u201codom\u201d data and publishes them.<\/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
This script publishes odometry data with ROS \u201codom\u201d topic. So, we can see the published data with the given command:<\/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
In this use case, we will just interested the given part of the data:<\/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
First of all, we generated a network called datapipeline for the architecture. The architecture consists of 4 services and each has a static IP address and uses the default port as the given below:<\/p>\n
Spark: 172.18.0.2
\nZookeeper: 172.18.0.3
\nKafka: 172.18.0.4
\nCassandra : 172.18.0.5<\/p><\/blockquote>\n
We use \u201cvolumes\u201d to import our scripts to containers.<\/p>\n
\u2757 You have to implement \u201c ..\/streamingProje:\/home\u201d part for your system.<\/em><\/p>\n
You can access the docker-compose<\/span><\/a>\u00a0\u00a0<\/span>and replace your configs.<\/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
Open your workspace folder which includes all files provided and run the given command as below.<\/p>\n
\n
# run docker-compose file\ndocker-compose up<\/span><\/pre>\n<\/blockquote>\n
You will have a view like:<\/p>\n
\n
\n
<\/picture>\"\"<\/a><\/div>\n<\/div>\n<\/figure>\n
After all, container is running, you can set up your environment.<\/p>\n
Prepare Kafka for Use Case<\/strong><\/pre>\n
First of all, we will create a new Kafka topic namely\u00a0odometry<\/em>\u00a0for ROS odom data using the given commands:<\/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
You will have a view like:<\/p>\n
\n
<\/picture>\"\"<\/a><\/div>\n<\/figure>\n
Prepare Cassandra for Use Case<\/strong><\/pre>\n
Initially, we will create a\u00a0keyspace<\/em>\u00a0and then a\u00a0topic<\/em>\u00a0in it using the given command:<\/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 The content of topic has to be the same as Spark schema: Be very careful here!<\/em><\/p>\n
4. Prepare Apache Spark structured streaming<\/strong><\/pre>\n
You are able to write analysis results to either console or Cassandra.<\/p>\n
(First Way) Prepare Apache Spark Structured Streaming Pipeline Kafka to Cassandra<\/strong><\/pre>\n
We will write a streaming script that reads\u00a0odometry<\/em>\u00a0topics from Kafka, analyzes them, and then write results to Cassandra. We will use\u00a0streamingKafka2Cassandra.py<\/a><\/span>\u00a0<\/span>to do it.<\/p>\n
First of all, we create a schema the same as we already defined in Cassandra.<\/p>\n
\u26a0\ufe0f The content of schema has to be the same as Casssandra table: Be very careful here!<\/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
Then, we create a Spark Session and specify our config here:<\/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
In order to read the Kafka stream, we use readStream() and specify Kafka configurations as the given below:<\/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
Since Kafka sends data as binary, first we need to convert the binary value to String using selectExpr() as the given below:<\/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
Although Apache Spark isn\u2019t capable of directly writing stream data to Cassandra yet (using writeStream()), we can do it by use foreachBatch() as the given below:<\/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
Finally, we got the given script\u00a0streamingKafka2Cassandra.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
There are a few differences between writing to the console and writing to Cassandra. We directly write stream to console. With writeStream() we can write stream data directly to the console.<\/p>\n
\n
df1.writeStream \\\n  .outputMode(\"update\") \\\n  .format(\"console\") \\\n  .option(\"truncate\", False) \\\n  .start() \\\n  .awaitTermination()<\/span><\/pre>\n<\/blockquote>\n
The rest of the process takes place in the same way as the previous one. Finally, we got the given script\u00a0streamingKafka2Console.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
If you are sure that all preparations are done, you can start a demo. You have to follow the given steps.<\/p>\n
Start ROS and publish odom data to Kafka.<\/strong><\/pre>\n