{"id":9601,"date":"2018-08-13T06:09:51","date_gmt":"2018-08-13T10:09:51","guid":{"rendered":"https:\/\/qxf2.com\/blog\/?p=9601"},"modified":"2018-09-02T09:43:42","modified_gmt":"2018-09-02T13:43:42","slug":"exploring-wallaroo-understanding-state-partitions-with-an-example","status":"publish","type":"post","link":"https:\/\/qxf2.com\/blog\/exploring-wallaroo-understanding-state-partitions-with-an-example\/","title":{"rendered":"Exploring Wallaroo: Understanding state partitions with an example"},"content":{"rendered":"

This post is aimed at Wallaroo<\/a> users who are looking to develop an intuitive understanding of state partitioning. Wallaroo is a framework that makes it easy to handle streaming data and write event processing applications quickly. Wallaroo already has a really nice example of state partitioning called alphabet_partitioned<\/code> but it is missing a final step illustrating how partitioning helps when there are multiple Wallaroo workers. In this post, we fill the gap. You will learn:
\na) how to run Wallaroo in multi-worker mode across multiple Docker containers
\nb) a simple testing trick to prove that state partitions help in parallelizing tasks<\/p>\n

\n

Overview<\/h3>\n

Wallaroo applications are built for processing millions of messages quickly. However, to make things illustrative, we are going to:
\na) control our input to have only 52 messages
\nb) introduce a sleep time in our computation to make the difference between the 1-worker and 2-worker experiments explicit<\/p>\n

We begin with a 1-worker setup and baseline the performance of the application. Then, when we move to a 2-worker setup, we can check if our latency numbers are (almost) halve since 2-workers are accessing the state in parallel. <\/p>\n

The rest of this guide is structured like this:<\/p>\n

    \n
  1. Setup<\/li>\n
  2. One worker experiment<\/li>\n
  3. Two worker experiment<\/li>\n
  4. An exercise<\/li>\n<\/ol>\n
    \n

    1. Setup<\/h3>\n

    We will rely on Wallaroo\u2019s existing alphabet_partitioned<\/a> application as a starting point to help us understand state partitioning better. To follow along, make sure you are setup with Wallaroo + Python<\/a> in Docker. The setup instructions are clear and easy to follow.<\/p>\n

    Once you are done with the Docker setup, start the Docker container for worker 1 by running the command:<\/p>\n

    \r\ndocker run --rm -it --privileged -p 4000:4000 \\\r\n-v \/tmp\/wallaroo-docker\/wallaroo-src:\/src\/wallaroo \\\r\n-v \/tmp\/wallaroo-docker\/python-virtualenv:\/src\/python-virtualenv \\\r\n--name wally \\\r\nwallaroo-labs-docker-wallaroolabs.bintray.io\/release\/wallaroo:0.5.0\r\n<\/pre>\n
    Your output should be similar to the image below:
    \n\"Wallaroo:<\/p>\n
    At the end of this step, you should have entered a Docker container called ‘wally’. Make sure to move to the alphabet_partitioned<\/code> directory by cd \/src\/wallaroo\/examples\/python\/alphabet_partitioned<\/code><\/p>\n
    Now, figure out the IP address of the container by running ifconfig<\/code>. For newbies, the IP address will usually be the ‘inet addr’ listed for the interface eth0<\/code>. Note down this IP address. For the rest of this post, we will refer to this IP address as $IP_Initializer<\/code>.<\/p>\n
    \n
    2. One worker experiment <\/h3>\n
    In this part of the experiment, you will have only one Docker container running Wallaroo. You will use this step to get a baseline for how long your application takes to process a given message set.<\/p>\n
    a. Create your own input file<\/strong>
    \nLet us handcraft a small input file and then use it to send a controlled set of messages. This will help us figure out when the experiment ends by simply looking at the output. <\/p>\n
    The alphabet_partitioned<\/code> application expects 9-byte messages:
    \na) a 4-byte header that has just the length of the payload
    \nb) a 1-byte alphabet
    \nc) a 4-byte number that represents the number of votes for the alphabet<\/p>\n
    To make things easy for us, we will design the file to have 52 messages. The first 26 messages will be an alphabet and the number 1. The second 26 messages will be an alphabet and the number 2.<\/p>\n
    Save the following code as (say) generate_binary_votes.py<\/code> in your \/src\/wallaroo\/examples\/python\/alphabet_partitioned directory.<\/p>\n
    \r\n\"\"\"\r\nA scratch script to produce a file similar to votes.msg for Wallaroo's alphabet_partitioned application\r\n\r\nDESIGN:\r\n1. Open a file fake_votes.msg in binary format\r\n2. Write 5 byte messages alphabet,vote \r\n3. Prepend a header with the number 5 (unsigned int) in it because all messages are 5 bytes\r\n4. Write this binary string to a file\r\n5. To make it easy to test for correctness, we will design the file to have only 52 messages \r\n6. The first 26 messages will be an alphabet and the number 1\r\n7. The second 26 messages will be an alphabet and the number 2\r\n\"\"\"\r\n\r\nimport string\r\nimport struct\r\n\r\ndef generate_alphabet_input(filename):\r\n    \"Generate a binary file with alphabets and votes\"\r\n    alphabets = list(string.ascii_lowercase)\r\n    with open(filename,'wb') as fp: \r\n        for i in range(1,3):\r\n            for alphabet in alphabets:\r\n                msg = struct.pack('>IsI',5,alphabet,i)\r\n                fp.write(msg)\r\n\r\n\r\n#----START OF SCRIPT\r\nif __name__=='__main__':\r\n    generate_alphabet_input('fake_votes.msg')\r\n<\/pre>\n
    Now, run the python script to produce a new 468 byte (=52*9) binary input file called fake_votes.msg<\/code>.<\/p>\n
    \r\npython generate_binary_votes.py\r\n<\/pre>\n
    At the end of this step, you should have a fake_votes.msg<\/code> file in your current directory.<\/p>\n
    b. Edit the application to add sleep time<\/strong>
    \nOpen alphabet_partitioned.py<\/code> using vi. Then, add two lines:<\/p>\n
    a) import time<\/code> along with the other imports
    \nb) time.sleep(0.5)<\/code> at the end of the update()<\/code> method for the class TotalVotes<\/code><\/p>\n
    By doing so, we are forcing the computation to consume ~0.5 seconds per message processed.<\/p>\n
    c. Start the receiver<\/strong>
    \nStart the receiver by executing:<\/p>\n
    \r\ndata_receiver --ponythreads=1 --ponynoblock   --listen $IP_Initializer:7002\r\n<\/pre>\n
    Note: $IP_Initializer<\/code> is the IP address you captured in the Setup section.<\/p>\n
    d. Start the application<\/strong>
    \nIn a new terminal prompt, enter the container<\/p>\n
    \r\ndocker exec -it wally env-setup\r\n<\/pre>\n
    Then, in the prompt of the Docker container, go to the alphabet_partitioned<\/code> application’s directory<\/p>\n
    \r\ncd wallaroo\/examples\/python\/alphabet_partitioned\/\r\n<\/pre>\n
    Start the application by executing:<\/p>\n
    \r\nmachida --application-module alphabet_partitioned \\\r\n  --in $IP_Initializer:7010 --out $IP_Initializer:7002 \\\r\n  --metrics $IP_Initializer:5001 --control $IP_Initializer:6000 \\\r\n  --data $IP_Initializer:6001 \\\r\n  --external $IP_Initializer:5050 --cluster-initializer\\\r\n  --ponythreads=1 --ponynoblock\r\n<\/pre>\n
    e. Start the sender<\/strong>
    \nIn a new terminal prompt, enter the container<\/p>\n
    \r\ndocker exec -it wally env-setup\r\n<\/pre>\n
    Then, in the prompt of the Docker container, go to the alphabet_partitioned<\/code> application’s directory<\/p>\n
    \r\ncd wallaroo\/examples\/python\/alphabet_partitioned\/\r\n<\/pre>\n
    IMPORTANT:<\/strong> Start a stopwatch as soon as you start the sender. We are sending 52 messages and each one will take ~0.5 seconds (our sleep time) to process. So we expect that the whole experiment will take about 26 seconds to complete. Start the sender by executing:<\/p>\n
    \r\nsender --host $IP_Initializer:7010 \\\r\n  --file fake_votes.msg --batch-size 26 --interval 10_000_000 \\\r\n  --messages 52 --binary --msg-size 9 --repeat --ponythreads=1 \\\r\n  --ponynoblock --no-write \r\n<\/pre>\n
    f. Time the app<\/strong>
    \nStop your timer as soon as you see ‘z=>3’ in the output. If things went well, this experiment took (approximately) 26 seconds.<\/p>\n
    g. Close the cluster<\/strong>
    \nIn the sender’s terminal prompt, execute <\/p>\n
    \r\ncluster_shutdown $IP_Initializer:5050\r\n<\/pre>\n
    At this point, the application should shut down. You can shut down the receiver by (Ctrl+c)-ing out of it. Leave the docker container running and all the terminals open since we are going to be using them in the two worker experiment too.<\/p>\n
    \n
    3. Two worker experiment<\/h3>\n
    We are going to introduce a second worker. Then, we give alphabet_partitioned<\/code> application the exact same input stream as before. If state partitioning is working right, we should observe that the time it takes to process the same 52 messages is now almost halved.  <\/p>\n
    a. Start the docker container of worker 2<\/strong>
    \nName the second docker container ‘roo’ (yeah, I know, imaginative!) and start it like this:<\/p>\n
    \r\ndocker run --rm -it --privileged -p 4001:4000 \\\r\n  -v \/tmp\/wallaroo-docker\/wallaroo-src:\/src\/wallaroo \\\r\n  -v \/tmp\/wallaroo-docker\/python-virtualenv:\/src\/python-virtualenv \\\r\n  --name roo \\\r\n  wallaroo-labs-docker-wallaroolabs.bintray.io\/release\/wallaroo:0.5.0\r\n<\/pre>\n
    Make sure to move to the alphabet_partitioned<\/code> directory by cd \/src\/wallaroo\/examples\/python\/alphabet_partitioned<\/code> <\/p>\n
    To be safe, examine the code in alphabet_partitioned.py<\/code> and confirm that your sleep statements are still there.<\/p>\n
    b. Start the receiver on worker 1<\/strong>
    \nIn the receiver prompt of the first worker repeat the command<\/p>\n
    \r\ndata_receiver --ponythreads=1 --ponynoblock   --listen $IP_Initializer:7002\r\n<\/pre>\n
    c. Start the application on worker 1<\/strong>
    \nWorker 1 (‘wally’) will be the cluster initializer. To start the application, run the following command on worker 1:<\/p>\n
    \r\nmachida --application-module alphabet_partitioned \\\r\n  --in $IP_Initializer:7010 --out $IP_Initializer:7002 \\\r\n  --metrics $IP_Initializer:5001 --control $IP_Initializer:6000 \\\r\n  --data $IP_Initializer:6001 --external $IP_Initializer:5050 \\\r\n  --cluster-initializer --worker-count 2\\\r\n  --ponythreads=1 --ponynoblock\r\n<\/pre>\n
    Notice we have added a worker-count<\/code> argument telling the application that it has two workers.<\/p>\n
    d. Start the application on worker 2<\/strong>
    \nUse the terminal prompt in step 3.a and run the following command on worker 2 (‘roo’):<\/p>\n
    \r\nmachida --application-module alphabet_partitioned \\\r\n  --in $IP_Initializer:7010 --out $IP_Initializer:7002 \\\r\n  --metrics $IP_Initializer:5001 --control $IP_Initializer:6000 \\\r\n  --ponythreads=1 --ponynoblock --name roo\r\n<\/pre>\n
    At the end of this step, you should see the output of the terminal claiming that the application is ready to work.<\/p>\n
    e. Start the sender on worker 1<\/strong>
    \nThis is very similar to step 2.e. Run the following command on worker 1.<\/p>\n
    IMPORTANT:<\/strong> Start a stopwatch as soon as you start the sender. We are sending 52 messages and each one will take ~0.5 seconds (our sleep time) to process. But 2 messages are going to be processed in parallel. So we expect that the whole experiment will take about 13 seconds (=52*0.5\/2) to complete.<\/p>\n
    In the sender’s terminal prompt, execute:<\/p>\n
    \r\nsender --host $IP_Initializer:7010 \\\r\n  --file fake_votes.msg --batch-size 26 --interval 10_000_000 \\\r\n  --messages 52 --binary --msg-size 9 --repeat --ponythreads=1 \\\r\n  --ponynoblock --no-write \r\n<\/pre>\n
    f. Time the app<\/strong>
    \nStop your timer as soon as you see ‘z=>3’ in the output of the receiver. If things went well, this experiment took (approximately) 13 seconds.<\/p>\n
    g. Close the cluster<\/strong>
    \nIn the sender’s terminal prompt, execute <\/p>\n
    \r\ncluster_shutdown $IP_Initializer:5050\r\n<\/pre>\n
    At this point, the application should shut down. You can shut down the receiver by (Ctrl+c)-ing out of it. <\/p>\n
    \n
    4. An exercise<\/h3>\n
    To further solidify your understanding through experimentation, try the same experiment by creating an input file with just a1 in binary format repeating 52 times. What do you expect? Is state-partitioning going to help reduce the total time taken to process the messages?<\/p>\n
    \n
    References<\/h3>\n
    1. Wallaroo Partitioning<\/a>
    \n2.     Stateful multi-steam processing in Python with Wallaroo<\/a><\/p>\n
    \n","protected":false},"excerpt":{"rendered":"
    This post is aimed at Wallaroo users who are looking to develop an intuitive understanding of state partitioning. Wallaroo is a framework that makes it easy to handle streaming data and write event processing applications quickly. Wallaroo already has a really nice example of state partitioning called alphabet_partitioned but it is missing a final step illustrating how partitioning helps when […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[154],"tags":[],"class_list":["post-9601","post","type-post","status-publish","format-standard","hentry","category-wallaroo"],"_links":{"self":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/9601","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/comments?post=9601"}],"version-history":[{"count":22,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/9601\/revisions"}],"predecessor-version":[{"id":9755,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/posts\/9601\/revisions\/9755"}],"wp:attachment":[{"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/media?parent=9601"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/categories?post=9601"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/qxf2.com\/blog\/wp-json\/wp\/v2\/tags?post=9601"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}