Distributed Computation for large data-data processing

I have a huge time series data and I want to do data processing using spark`s parallel processing/distributed computation. The requirement is looking at the data row by row to determine the groups as specified below under desired result sections, I can't really get spark to distribute this without some kind of coordination between the executors

t- timeseries datetime sample, lat-latitude, long-longitude 

For instance : Taking a small part of sample data-set for explaining the case

t   lat long 0   27  28 5   27  28 10  27  28 15  29  49 20  29  49 25  27  28 30  27  28  

Desired Output should be :

Lat-long    interval (27,28) (0,10) (29,49) (15,20) (27,28) (25,30) 

I am able to get the desired result using this piece of code

val spark = SparkSession.builder().master("local").getOrCreate()  import spark.implicits._   val df = Seq(   (0, 27,28),   (5, 27,28),   (10, 27,28),   (15, 26,49),   (20, 26,49),   (25, 27,28),   (30, 27,28) ).toDF("t", "lat","long")  val dfGrouped = df .withColumn("lat-long", struct($"lat", $"long"))  val wAll = Window.partitionBy().orderBy($"t".asc)  dfGrouped.withColumn("lag", lag("lat-long", 1, null).over(wAll)) .orderBy(asc("t")).withColumn("detector", when($"lat-long" === $"lag", 0)     .otherwise(1)).withColumn("runningTotal", sum("detector").over(wAll)) .groupBy("runningTotal", "lat-long").agg(struct(min("t"), max("t")).as("interval")) .drop("runningTotal").show } 

But what If the data gets into two executors then the data will be like

Data in executor 1 :

t   lat long 0   27  28 5   27  28 10  27  28 15  29  49 20  29  49 25  27  28 

Data in executor 2 :

t   lat long 30   27  28 

How should I get the desired output for large amount of data.There must be smarter ways to do this ,distributing this with some kind of coordination between the executors so as to get that result.

Please guide me through a right direction,I have researched about the same but not being able to land up to a solution.

PS: This just a sample example.

1 Answers

Answers 1

You can address this with a UDAF. First of all, you could add one column thats represent the t column partitioned in a number of executor you have. Something like executorIndex = t % ((max(t) - min(t)) / numExecutors).

Then you can apply your UDAF grouping by executorIndex.

Your UDAF need store a Map with a String key (for example) thats represents one lat and long pair, and a int[] thats represents the maxT and the minT for this lat-long key.

Please ask if you need more extensive explanation.

Hope this help...

PS: I'm suming that there are some time relation between same lat and long, something normal if your are tracking some movement...

Read More

Tensorflow can't detect GPU when invoked by Ray worker

When I try the following code sample for using Tensorflow with Ray, Tensorflow fails to detect the GPU's on my machine when invoked by the "remote" worker but it does find the GPU's when invoked "locally". I put "remote" and "locally" in scare quotes because everything is running on my desktop which has two GPU's and is running Ubuntu 16.04 and I installed Tensorflow using the tensorflow-gpu Anaconda package.

The local_network seems to be responsible for these messages in the logs:

2018-01-26 17:24:33.149634: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1045] Creating TensorFlow device (/gpu:0) -> (device: 0, name: Quadro M5000, pci bus id: 0000:03:00.0) 2018-01-26 17:24:33.149642: I tensorflow/core/common_runtime/gpu/gpu_device.cc:1045] Creating TensorFlow device (/gpu:1) -> (device: 1, name: Quadro M5000, pci bus id: 0000:04:00.0) 

And the remote_network seems to be responsible for this message:

2018-01-26 17:24:34.309270: E tensorflow/stream_executor/cuda/cuda_driver.cc:406] failed call to cuInit: CUDA_ERROR_NO_DEVICE 

Why is Tensorflow able to detect the GPU in one case but not the other?

import tensorflow as tf import numpy as np import ray  ray.init()  BATCH_SIZE = 100 NUM_BATCHES = 1 NUM_ITERS = 201  class Network(object):     def __init__(self, x, y):         # Seed TensorFlow to make the script deterministic.         tf.set_random_seed(0)         # Define the inputs.         x_data = tf.constant(x, dtype=tf.float32)         y_data = tf.constant(y, dtype=tf.float32)         # Define the weights and computation.         w = tf.Variable(tf.random_uniform([1], -1.0, 1.0))         b = tf.Variable(tf.zeros([1]))         y = w * x_data + b         # Define the loss.         self.loss = tf.reduce_mean(tf.square(y - y_data))         optimizer = tf.train.GradientDescentOptimizer(0.5)         self.grads = optimizer.compute_gradients(self.loss)         self.train = optimizer.apply_gradients(self.grads)         # Define the weight initializer and session.         init = tf.global_variables_initializer()         self.sess = tf.Session()         # Additional code for setting and getting the weights         self.variables = ray.experimental.TensorFlowVariables(self.loss, self.sess)         # Return all of the data needed to use the network.         self.sess.run(init)      # Define a remote function that trains the network for one step and returns the     # new weights.     def step(self, weights):         # Set the weights in the network.         self.variables.set_weights(weights)         # Do one step of training. We only need the actual gradients so we filter over the list.         actual_grads = self.sess.run([grad[0] for grad in self.grads])         return actual_grads      def get_weights(self):         return self.variables.get_weights()  # Define a remote function for generating fake data. @ray.remote(num_return_vals=2) def generate_fake_x_y_data(num_data, seed=0):     # Seed numpy to make the script deterministic.     np.random.seed(seed)     x = np.random.rand(num_data)     y = x * 0.1 + 0.3     return x, y  # Generate some training data. batch_ids = [generate_fake_x_y_data.remote(BATCH_SIZE, seed=i) for i in range(NUM_BATCHES)] x_ids = [x_id for x_id, y_id in batch_ids] y_ids = [y_id for x_id, y_id in batch_ids] # Generate some test data. x_test, y_test = ray.get(generate_fake_x_y_data.remote(BATCH_SIZE, seed=NUM_BATCHES))  # Create actors to store the networks. remote_network = ray.remote(Network) actor_list = [remote_network.remote(x_ids[i], y_ids[i]) for i in range(NUM_BATCHES)] local_network = Network(x_test, y_test)  # Get initial weights of local network. weights = local_network.get_weights()  # Do some steps of training. for iteration in range(NUM_ITERS):     # Put the weights in the object store. This is optional. We could instead pass     # the variable weights directly into step.remote, in which case it would be     # placed in the object store under the hood. However, in that case multiple     # copies of the weights would be put in the object store, so this approach is     # more efficient.     weights_id = ray.put(weights)     # Call the remote function multiple times in parallel.     gradients_ids = [actor.step.remote(weights_id) for actor in actor_list]     # Get all of the weights.     gradients_list = ray.get(gradients_ids)      # Take the mean of the different gradients. Each element of gradients_list is a list     # of gradients, and we want to take the mean of each one.     mean_grads = [sum([gradients[i] for gradients in gradients_list]) / len(gradients_list) for i in range(len(gradients_list[0]))]      feed_dict = {grad[0]: mean_grad for (grad, mean_grad) in zip(local_network.grads, mean_grads)}     local_network.sess.run(local_network.train, feed_dict=feed_dict)     weights = local_network.get_weights()      # Print the current weights. They should converge to roughly to the values 0.1     # and 0.3 used in generate_fake_x_y_data.     if iteration % 20 == 0:         print("Iteration {}: weights are {}".format(iteration, weights)) 

1 Answers

Answers 1

The GPUs are cut off by ray.remote decorator itself. From its source code:

def remote(*args, **kwargs):     ...     num_cpus = kwargs["num_cpus"] if "num_cpus" in kwargs else 1     num_gpus = kwargs["num_gpus"] if "num_gpus" in kwargs else 0  # !!!     ... 

So the following call effectively sets num_gpus=0:

remote_network = ray.remote(Network) 

Ray API is a bit strange, and you can't simply say ray.remote(Network, num_gpus=2) (though that's exactly what you want). Here's what I did and it seems to work on my machine:

ray.init(num_gpus=2)  ...  @ray.remote(num_gpus=2) class RemoteNetwork(Network):     pass  actor_list = [RemoteNetwork.remote(x_ids[i],y_ids[i]) for i in range(NUM_BATCHES)] 

Read More

What kinds of out-of-band failures am I forgetting to test?

I came across an amazing presentation years ago (which of course I can't find) that listed a bunch of kinds of failures for remote services that people usually don't test for.

In addition to timeout, 4xx, 5xx, etc, it listed things like:

• connection closes after 10 bytes of data
• returns contents of www.google.com
• returns contents of /dev/random
• returns contents of /etc/passwd
• returns correctly-formatted unicode chinese text
• returns ansi color control characters
• returns an incorrect content-type, labeled correctly (You requested Content-Type: application/json, I send back Content-Type: application/jpeg)
• returns one byte of data every 29 seconds

What are some types of "out-of-band failures" you've encountered that developers don't usually (but should) test for?

(extra bonus points if you can find the original presentation)

2 Answers

Answers 1

The ones you listed are great; I'd love to see the original presentation if you dig it up! A couple other favorites:

• A "valid" response with a couple bits flipped
• A "valid" response with extra data you weren't expecting ({"result": 123, "extraStuff": {...}}) to simulate upgrades to the remote side
• A syntactically-valid response that never ends ({"results":["lol", "lol", "lol", ..., or just a bunch of whitespace)

Answers 2

Low-frequency failures. In other words, test that some response is correct not just once, but every time out of a thousand tries. You'll get random Internet breakage if you're going over a network, but you might expose some process is stochastic when you thought it was fixed.

Read More

How to handle file paths in distributed environment

I'm working on setting up a distributed celery environment to do OCR on PDF files. I have about 3M PDFs and OCR is CPU-bound so the idea is to create a cluster of servers to process the OCR.

As I'm writing my task, I've got something like this:

@app.task def do_ocr(pk, file_path):     content = run_tesseract_command(file_path)     item = Document.objects.get(pk=pk)     item.content = ocr_content     item.save() 

The question I have what the best way is to make the file_path work in a distributed environment. How do people usually handle this? Right now all my files simply live in a simple directory on one of our servers.

3 Answers

Answers 1

Well, there are multiple ways to handle it, but let's stick to one of the simpliest one:

• since you'd like to process big amount of files using multiple servers, my first suggestion would be to use the same OS in each server, so you won't have to worry about cross-platform compatibility
• using the word 'cluster' indicates that all of those servers should know their mutual state - it adds complexity, try to switch to the farm of stateless workers (by 'stateless' I mean "not knowing about other's" as they should be aware of at least their own state, e.g.: IDLE, IN_PROGRESS, QUEUE_FULL or more if needed)
• for the file list processing part you could use pull or push model:
  • push model could be easily implemented by a simple app that crawls the files and dispatches them (e.g.: over SCP, FTP, whatever) to a set of available servers; servers can monitor their local directories for changes and pick up new files to process; it's also very easy to scale - just spin up more servers and update the push client (even in runtime); the only limit is your push client's performance
  • pull model is a little bit more tricky, cause you have to handle more complexity; having a set of servers implicates having a proper starting index per node and offset - it will make error handling more difficult, plus, it doesn't scale easily (imagine adding twice as more servers to speedup the processing and updating indices and offsets properly on each node.. seems like an error-prone solution)
• I assume that the network traffic isn't a big concern - having 3M files to process will generate it somewhere, one way or the other..
• collecting/storing the results is a different ballpark, but here the list of possible solutions is limitless

Answers 2

If your are in linux environment the easiest way is mount a remote filesystem, using sshfs, in the /mnt folder foreach node in cluster. Then you can pass the node name to do_ocr function and work as all data is local to current node

For example, your cluster has N nodes named: node1, ... ,nodeN
Let's configure node1, foreach node mount remote filesystem. Here's a sample node1's /etc/fstab file

sshfs#user@node2:/var/your/app/pdfs    /mnt/node2 fuse    port=<port>,defaults,user,noauto,uid=1000,gid=1000        0       0 .... sshfs#user@nodeN:/var/your/app/pdfs    /mnt/nodeN fuse    port=<port>,defaults,user,noauto,uid=1000,gid=1000        0       0 

In current node (node1) create a symlink named as current server pointing to pdf's path

ln -s /var/your/app/pdfs node1 

Your mnt folder should contain remote's filesystem and a symlink

user@node1:/mnt$ ls -lsa 0 lrwxrwxrwx  1 user user      16 apr 12  2016 node1 -> /var/your/app/pdfs 0 lrwxrwxrwx  1 user user      16 apr 12  2016 node2 ... 0 lrwxrwxrwx  1 user user      16 apr 12  2016 nodeN 

Then your function should look like this:

import os MOUNT_POINT = '/mtn' @app.task def do_ocr(pk, node_name, file_path):     content = run_tesseract_command(os.path.join(MOUNT_POINT,node_name,file_path))     item = Document.objects.get(pk=pk)     item.content = ocr_content     item.save() 

It works like all files are in the current machine but there's remote-logic working for you transparently

Answers 3

Since I miss a lot of your architecture details and your application specifics, you can take this answer as a guiding answer rather than a strict one. You can take this approach, in the following order:

1- deploy an internal file server that stores all the files in one place and serve them

Example:

http://interanal-ip-address/storage/filenameA.pdf

http://interanal-ip-address/storage/filenameB.pdf

http://interanal-ip-address/storage/filenameC.pdf

and so on ...

2- Install/Deploy Redis

3- Create an upload client/service/process that takes the files you want to upload and pass them to the above storage location (/storage/), so your files will be available once they are uploaded, at the same time push the full file path URL to a predefined Redis List/Queue (build on linked lists data structure), like this: http://internal-ip-address/storage/filenameA.pdf

You can get more details here about LPUSH and RPOP under Redis Lists here: http://redis.io/topics/data-types-intro

Examples:

1. A file upload form, that stores the files directly to storage area
2. A file upload utility/command-line/background-process, that you can create it yourself or use some existing tool to upload files to the storage location, that gets the files from specific location, be it a web address or some other server that has your files

4- Now we come to your celery workers, each one of your workers should pull (RPOP) one of the files URLs from Redis queue, download the file from your internal file server (we built in first step), and do the required processing on the way you wanted it to be.

An important thing to note from Redis documentation:

Lists have a special feature that make them suitable to implement queues, and in general as a building block for inter process communication systems: blocking operations.

However it is possible that sometimes the list is empty and there is nothing to process, so RPOP just returns NULL. In this case a consumer is forced to wait some time and retry again with RPOP. This is called polling, and is not a good idea in this context because it has several drawbacks

So Redis implements commands called BRPOP and BLPOP which are versions of RPOP and LPOP able to block if the list is empty: they'll return to the caller only when a new element is added to the list, or when a user-specified timeout is reached.

Let me know if that answers your question.

Things to keep in mind

• You can add as many workers as you want since this solution is very scalable, and your only bottleneck is Redis server, which you can make cluster of and persist your queue in case of power outage or server crash

• You can replace redis with RabbitMQ, Beanstalk, Kafka, or any other queuing/messaging system, but Redis has ben nominated in this race due to simplicity and meriad of features introduced out of the box.

Read More