{"id":4863,"date":"2020-11-06T13:20:02","date_gmt":"2020-11-06T13:20:02","guid":{"rendered":"https:\/\/machinelearningplus.com\/?p=4863"},"modified":"2022-04-04T07:43:19","modified_gmt":"2022-04-04T07:43:19","slug":"dask-tutorial","status":"publish","type":"post","link":"https:\/\/machinelearningplus.com\/python\/dask-tutorial\/","title":{"rendered":"Dask – How to handle large dataframes in python using parallel computing"},"content":{"rendered":"

Dask provides efficient parallelization for data analytics in python. Dask Dataframes allows you to work with large datasets for both data manipulation and building ML models with only minimal code changes. It is open source and works well with python libraries like NumPy, scikit-learn, etc. Let’s understand how to use Dask with hands-on examples.<\/em><\/p>\n

\"\"Dask – How to handle large data in python using parallel computing<\/img><\/p>\n

Contents<\/h2>\n
    \n
  1. Why do you need Dask?<\/a><\/li>\n
  2. What is Dask?<\/a><\/li>\n
  3. Quickly about Parallel Processing<\/a><\/li>\n
  4. How to implement Parallel Processing with Dask<\/a><\/li>\n
  5. What does dask.delayed do?<\/a><\/li>\n
  6. Example: Parallelizing a for loop with Dask<\/a><\/li>\n
  7. How to use Dask DataFrames<\/a><\/li>\n
  8. How is dask.dataframe different from pandas.dataframe?<\/a><\/li>\n
  9. Introduction to Dask Bags<\/a><\/li>\n
  10. How to use Dask Bag for various operations?<\/a><\/li>\n
  11. Distributed computing with Dask – Hands-on Example<\/a><\/li>\n<\/ol>\n

    Why do you need Dask?<\/h2>\n

    Python packages like numpy, pandas, sklearn, seaborn etc. makes the data manipulation and ML tasks very convenient. For most data analysis tasks, the python pandas<\/a> package is good enough. You can do all sorts of data manipulation and is compatible for building ML models.<\/p>\n

    But, as your data gets bigger, bigger than what you can fit in the RAM, pandas won’t be sufficient.<\/p>\n

    This is a very common problem.<\/p>\n

    You may use Spark or Hadoop to solve this. But, these are not python environments. This stops you from using numpy, sklearn, pandas, tensorflow, and all the commonly used Python libraries for ML.<\/p>\n

    Is there a solution for this?<\/p>\n

    Yes! This is where Dask comes in.<\/p>\n

    What is Dask?<\/h2>\n

    Dask is a open-source library that provides advanced parallelization for analytics<\/strong>, especially when you are working with large data.<\/p>\n

    It is built to help you improve code performance and scale-up without having to re-write your entire code. The good thing is, you can use all your favorite python libraries as Dask is built in coordination with numpy, scikit-learn, scikit-image, pandas, xgboost, RAPIDS and others.<\/p>\n

    That means you can now use Dask to not only speed up computations on datasets using parallel processing, but also build ML models using scikit-learn, XGBoost on much larger datasets.<\/p>\n

    You can use it to scale your python code for data analysis. If you think, this sounds a bit complicated to implement, just read on.<\/p>\n

    Related Post:<\/strong> Basics of python parallel processing with multiprocessing, clearly explained<\/a>.<\/p>\n

    Quickly about Parallel Processing<\/h2>\n

    So, What is Parallel Processing?<\/p>\n

    Parallel processing refers to executing multiple tasks at the same time, using multiple processors in the same machine.<\/p>\n

    Generally, the code is executed in sequence, one task at a time. But, let’s suppose, you have a complex code that takes a long time to run, but mostly the code logics are independent, that is, no data or logic dependency on each other. This is the case for most matrix operations.<\/p>\n

    So, instead of waiting for the previous task to complete, we compute multiple steps simultaneously at the same time<\/strong>. This lets you take advantage of the available processing power, which is the case in most modern computers, thereby reducing the total time taken.<\/p>\n

    Dask is designed to do this efficiently on datasets with minimal learning curve. Let’s see how.<\/p>\n

    How to implement Parallel Processing with Dask<\/h2>\n

    A very simple way is to use the dask.delayed<\/code> decorator to implement parallel processing. Let me explain it through an example.<\/p>\n

    Consider the below code snippet.<\/p>\n

    from time import sleep\n\ndef apply_discount(x):\n  sleep(1)\n  x=x-0.2*x\n  return x\n\ndef get_total(a,b):\n  sleep(1)\n  return a+b\n\n\ndef get_total_price(x,y):\n  sleep(1)\n  a=apply_discount(x)\n  b=apply_discount(y)\n  get_total(a,b)\n<\/code><\/pre>\n
    Given a number, the above code simply applies a 20 percent discount on price and then add them. I’ve inserted a sleep<\/code> function explicitly so both the functions take 1sec to run. This is a small code that will run quickly, but I have chosen this to demonstrate for beginners.<\/p>\n
    %%time\n# This takes three seconds to run because we call each\n# function sequentially, one after the other\n\nx = apply_discount(100)\ny = apply_discount(200)\nz = get_total_price(x,y)\n<\/code><\/pre>\n
    CPU times: user 859 \u00b5s, sys: 202 \u00b5s, total: 1.06 ms\nWall time: 6.01 s\n<\/code><\/pre>\n
    I have recorded the time taken for this execution using %%time<\/code> as shown. You can observe that time taken is 6.01 seconds, when it is executed sequentially. Now, let’s see how to use dask.delayed<\/code> to reduce this time.<\/p>\n
    # Import dask and and dask.delayed\nimport dask\nfrom dask import delayed\n<\/code><\/pre>\n
    Now, you can transform the functions apply_discount()<\/code> and get_total_price()<\/code>. You can use  delayed()<\/code> function to wrap the function calls that you want to turn into tasks.<\/p>\n
    # Wrapping the function calls using dask.delayed\nx = delayed(apply_discount)(100)\ny = delayed(apply_discount)(200)\nz = delayed(get_total_price)(x, y)\n<\/code><\/pre>\n
    What does dask.delayed do?<\/h2>\n
    It creates a delayed<\/code> object, that keeps track of all the functions to call and the arguments to pass to it. Basically, it builds a task graph that explains the entire computation.  It helps to spot opportunities for parallel execution.<\/p>\n
    So, the z<\/code> object created in the above code is a delayed object OR “lazy object” which has all information for executing the logic. You can see the optimal task graph created by dask by calling the visualize()<\/code> function.<\/p>\n
    z.visualize()\n<\/code><\/pre>\n
    \"\"<\/p>\n
    Clearly from the above image, you can see there are two instances of apply_discount()<\/code> function called in parallel. This is an opportunity to save time and processing power by executing them simultaneously.<\/p>\n
    Up until now, only the logic to compute the output, that is the task graph is computed. To actually execute it, let’s call the compute()<\/code> method of z<\/code>.<\/p>\n
    %%time\nz.compute()\n<\/code><\/pre>\n
    CPU times: user 6.33 ms, sys: 1.35 ms, total: 7.68 ms\nWall time: 5.01 s\n<\/code><\/pre>\n
    Though it’s just 1 sec, the total time taken has reduced. This is the basic concept of parallel computing. Dask makes it very convenient.<\/p>\n
    Let’s now look at more useful examples.<\/p>\n
    Example: Parallelizing a for loop with Dask<\/h2>\n
    In the previous section, you understood how dask.delayed<\/code> works. Now, let’s see how to do parallel computing in a for-loop<\/code>.<\/p>\n
    Consider the below code.<\/p>\n
    You have a for-loop<\/code>, where for each element a series of functions is called.<\/p>\n
    In this case, there is a lot of opportunity for parallel computing. Again, we wrap the function calls with delayed()<\/code>, to get the parallel computing task graph.<\/p>\n
    # Functions to perform mathematics operations\ndef square(x):\n    return x*x\n\ndef double(x):\n    return x*2\n\ndef add(x, y):\n    return x + y\n\n# For loop that calls the above functions for each data\noutput = []\nfor i in range(6):\n    a = delayed(square)(i)\n    b = delayed(double)(i)\n    c = delayed(add)(a, b)\n    output.append(c)\n\ntotal = dask.delayed(sum)(output)\n\n# Visualizing the task graph for the problem\ntotal.visualize()\n<\/code><\/pre>\n
    For this case, the total<\/code> variable is the lazy object. Let’s visualize the task graph using total.visualize()<\/code>.<\/p>\n
    \"\"<\/p>\n
    You can see from above that as problems get more complex, so here, parallel computing becomes more useful and necessary.<\/p>\n
    Now, wrapping every function call inside delayed()<\/code> becomes laborious. But then, the delayed<\/code> function is actually a Decorator<\/strong>. So, you can just add the @delayed<\/code> decorator before the function definitions as shown below. This reduces the number of code changes.<\/p>\n
    # Using delayed as a decorator to achieve parallel computing.\n\n@delayed\ndef square(x):\n    return x*x\n\n@delayed\ndef double(x):\n    return x*2\n\n@delayed\ndef add(x, y):\n    return x + y\n\n# No change has to be done in function calls\noutput = []\nfor i in range(6):\n    a = square(i)\n    b = double(i)\n    c = add(a, b)\n    output.append(c)\n\ntotal = dask.delayed(sum)(output)\ntotal.visualize()\n<\/code><\/pre>\n
    \"Example<\/p>\n
    As expected, you get the same output.<\/p>\n
    So you can use delayed<\/code> as a decorator as is and it will parallelize a for-loop as well. Isn’t that awesome?<\/p>\n
    Dask DataFrames – How to use them?<\/h2>\n
    You saw how Dask helps to overcome the problem of long execution and training time. Another important problem we discussed was the larger-than-memory datasets<\/strong>.<\/p>\n
    The commonly used library for working with datasets is Pandas. But, many real-life ML problems have datasets that are larger than your RAM memory!<\/p>\n
    In these cases, Dask Dataframes is useful. You can simply import the dataset as dask.dataframe<\/code> instead, which you can later convert to a pandas dataframe after necessary wrangling\/calculations are done.<\/p>\n
    How is dask.dataframe different from pandas.dataframe?<\/h2>\n
    A Dask DataFrame is a large parallel DataFrame composed of many smaller Pandas DataFrames, split along the index. One Dask DataFrame is comprised of many in-memory pandas DataFrames separated along with the index.<\/p>\n
    These Pandas DataFrames may live on disk for larger-than-memory computing on a single machine, or on many different machines in a cluster. One Dask DataFrame operation triggers many operations on the constituent Pandas DataFrames.<\/p>\n
    \"Dask<\/p>\n
    The Dask Dataframe interface is very similar to Pandas, so as to ensure familiarity for pandas users. There are some differences which we shall see.<\/p>\n
    For understanding the interface, let’s start with a default dataset provided by Dask. I have used dask.datasets.timeseries()<\/code> function, which can create time-series from random data.<\/p>\n
    import dask\nimport dask.dataframe as dd\ndata_frame = dask.datasets.timeseries()\n<\/code><\/pre>\n
    The data_frame<\/code> variable is now our dask dataframe. In padas, if you the variable, it’ll print a shortlist of contents. Let’s see what happens in Dask.<\/p>\n
    data_frame\n<\/code><\/pre>\n
    You can see that only the structure is there, no data has been printed. It’s because Dask Dataframes are lazy and do not perform operations unless necessary. You can use the head()<\/code> method to visualize data<\/p>\n
    data_frame.head()\n<\/code><\/pre>\n
    \"data<\/p>\n
    Now, let’s just perform a few basic operations which are expected from pandas using dask dataframe now. One of the most standard operations is to groupby()<\/code>.<\/p>\n
    # Applying groupby operation\ndf = data_frame.groupby('name').y.std()\ndf\n<\/code><\/pre>\n
    Dask Series Structure:\nnpartitions=1\n    float64\n        ...\nName: y, dtype: float64\nDask Name: sqrt, 67 tasks\n<\/code><\/pre>\n
    If you want the results then you can call compute()<\/code> function as shown below.<\/p>\n
    df.compute()\n<\/code><\/pre>\n
    name\nAlice       0.575963\nBob         0.576803\nCharlie     0.577633\nDan         0.578868\nEdith       0.577293\nFrank       0.577018\nGeorge      0.576834\nHannah      0.577177\nIngrid      0.578378\nJerry       0.577362\nKevin       0.577626\nLaura       0.577829\nMichael     0.576828\nNorbert     0.576417\nOliver      0.576665\nPatricia    0.577810\nQuinn       0.578222\nRay         0.577239\nSarah       0.577831\nTim         0.578482\nUrsula      0.576405\nVictor      0.577622\nWendy       0.577442\nXavier      0.578316\nYvonne      0.577285\nZelda       0.576796\nName: y, dtype: float64\n<\/code><\/pre>\n
    Sometimes the original dataframe may be larger than RAM, so you would have loaded it as Dask dataframe. After performing some operations, you might get a smaller dataframe which you would like to have in Pandas. You can easily convert a Dask dataframe into a Pandas dataframe by storing df.compute()<\/code>.<\/p>\n
    The compute()<\/code> function turns a lazy Dask collection into its in-memory equivalent (in this case pandas dataframe). You can verify this with type()<\/code> function as shown below.<\/p>\n
    # Converting dask dataframe into pandas dataframe\nresult_df=df.compute()\ntype(result_df)\n<\/code><\/pre>\n
    pandas.core.series.Series\n<\/code><\/pre>\n
    Another useful feature is the persist()<\/code> function of dask dataframe.<\/p>\n
    So, what does persist()<\/code> function do?<\/strong><\/p>\n
    This function turns a lazy Dask collection into a Dask collection with the same metadata. The difference is earlier the results were not computed, it just had the information. Now, the results are fully computed or actively computing in the background.<\/p>\n
    This function is particularly useful when using distributed systems, because the results will be kept in distributed memory, rather than returned to the local process as with compute.<\/p>\n
    # Calling the persist function of dask dataframe\ndf = df.persist()\n<\/code><\/pre>\n
    The majority of the normal operations have a similar syntax to theta of pandas. Just that here for actually computing results at a point, you will have to call the compute()<\/code> function. Below are a few examples that demonstrate the similarity of Dask with Pandas API.<\/p>\n
    df.loc['2000-01-05']\n<\/code><\/pre>\n
    Dask Series Structure:\nnpartitions=1\n    float64\n\nName: y, dtype: float64\nDask Name: try_loc, 2 tasks\n<\/code><\/pre>\n
    Now using compute()<\/code> on this materializes it.<\/p>\n
    %time \ndf.loc['2000-01-05'].compute()\n<\/code><\/pre>\n
    CPU times: user 3.03 ms, sys: 0 ns, total: 3.03 ms\nWall time: 2.87 ms\n\nSeries([], Name: y, dtype: float64)\n<\/code><\/pre>\n
    Introduction to Dask Bags<\/h2>\n
    In many cases, the raw input has a lot of messy data that needs processing. The messy data is often processed and represented as a sequence of arbitrary inputs. Usually, they are processed in form of lists, dicts, sets, etc. A common problem is when they take up a lot of storage and iterating through them takes time.<\/p>\n
    Is there a way to optimize data processing at raw-level?<\/p>\n
    Yes! The answer is Dask Bags.<\/p>\n
    What are Dask Bags?<\/strong><\/p>\n
    Dask.bag<\/code> is a high-level Dask collection used as an alternative for the regular python lists, etc. The main difference is Dask Bags are lazy and distributed.<\/p>\n
    Dask Bag implements operations like map, filter, fold, and groupby on collections of generic Python objects. We prefer Dask bags because it provides the best optimization.<\/p>\n
    What are the advantages of using Dask bags ?<\/strong><\/p>\n
      \n
    1. It lets you process large volumes of data in a small space, just like toolz<\/code>.<\/li>\n
    2. Dask bags follow parallel computing. The data is split up, allowing multiple cores or machines to execute in parallel<\/li>\n
    3. The execution part usually consists of running many iterations. In these iterations,  data is processed lazily in the case of Dask bag. It allows for smooth execution. <\/li>\n<\/ol>\n
      Because of the above points, Dask bags are often used on unstructured or semi-structured data like text data, log files, JSON records, etc.<\/p>\n
      How to create Dask Bags?<\/h3>\n
      Dask provides you different ways to create a bag from various python objects. Let’s look at each method with an example.<\/p>\n
      Method 1. Create a bag from a sequence :<\/p>\n
      You can create a dask Bag from Python sequence using the dask.bag.from_sequence()<\/code> function.
      \n  The parameters are :
      \n  seq<\/code>: The sequence of elements you wish to input<\/p>\n
      partition_size<\/code>: An integer to denote the size of each partition<\/p>\n
      The below example shows how to create a bag from a list. After creating, you can perform a wide variety of functions on the bag. For, example, visualize()<\/code> function returns a dot graph to represent the bag.<\/p>\n
      bag_1 = dask.bag.from_sequence(['Haritha', 'keerthi', 'Newton','Swetha','Sinduja'], partition_size=2)\nbag_1.visualize()\n<\/code><\/pre>\n
      \"\"<\/p>\n
      Method 2. Create bag from  dask Delayed objects :<\/p>\n
       You can create a dask Bag from dask Delayed objects using the `dask.bag.from_delayed()` function. The parameters is `values`. It refers to the list of dask Delayed objects you wish to input\n<\/code><\/pre>\n
      # Creating dask delayed objects\nx, y, z =[delayed(load_sequence_from_file)(fn) for fn in filenames]\n\n# Creating a bask using from_delayed()\nb = dask.bag.from_delayed([x, y, z])\n<\/code><\/pre>\n
      Method 3. Create a bag from text files:<\/p>\n
       You can create a dask Bag from a text file using the `dask.bag.read_text()` function. The main parameters are : \n\n `urlpath`: You can pass the path of the desired text file here.\n\n `blocksize`: In case the files are large, you can provide an option to cut them using this parameter\n\n `collection`: It is a boolean value parameter. The function will return `dask.bag` if True. Otherwise will return a list of delayed values.\n\n `include_path`: It is again a boolean parameter that decides\n<\/code><\/pre>\n
      whether or not to include the path in the bag. If true, elements are tuples of (line, path). By default, it is set to False.<\/p>\n
       The below example shows how to create a bag from a textfile\n<\/code><\/pre>\n
      b = read_text('myfiles.1.txt')  # doctest: +SKIP\nb = read_text('myfiles.*.txt')  \n\n# Parallelize a large file by providing the number of uncompressed bytes to load into each partition\nb = read_text('largefile.txt', blocksize='10MB')\n\n# Get file paths of the bag by setting include_path=True\nb = read_text('myfiles.*.txt', include_path=True)\n<\/code><\/pre>\n
      Method 4. Create a Dask bag from url:<\/p>\n
      You can create a dask Bag from a URL using the dask.bag.from_url()<\/code> function. You just need to input the url path, no other parameter<\/p>\n
       The below example shows how to create a bag from a url\n<\/code><\/pre>\n
      a = dask.bag.from_url('http:\/\/raw.githubusercontent.com\/dask\/dask\/master\/README.rst',) \na.npartitions  \n<\/code><\/pre>\n
      b = dask.bag.from_url(['http:\/\/github.com', 'http:\/\/google.com']) \nb.npartitions\n<\/code><\/pre>\n
      How to use Dask Bag for various operations?<\/h2>\n
      The previous section told us the different ways of creating dask bags. Now that you are familiar with the idea, let’s see how to perform various processing operations.<\/p>\n
      For our purpose,let’s create a dask bag using the make_people()<\/code> function available in dask.datasets<\/code>. This function make_people()<\/code> makes a Dask Bag with dictionary records of randomly generated people. To do this, it requires the library mimesis<\/code> to generate records. So, you have to install that too.<\/p>\n
      !pip install mimesis\n!pip install dask==1.0.0 distributed'>=1.21.6,<2.0.0'\nimport dask\nimport json\nimport os\n\n # Create data\/ directory\nos.makedirs('\/content\/my_data', exist_ok=True)       \n\n\nmy_bag = dask.datasets.make_people()\nmy_bag\n<\/code><\/pre>\n
      dask.bag\n<\/code><\/pre>\n
      The above code has successfully created a dask bag my_bag<\/code> that stores information. You can also see that the number of partitions is 10. Sometimes, you may need to write the data into a disk.<\/p>\n
      How to write the data in my_bag<\/code> (of 10 partitions) into 10 JSON files and store them?<\/h3>\n
      In situations like these, the dask.bag.map()<\/code> is pretty useful.dask.
      \nThe syntax is : bag.map(func, *args, **kwargs)<\/code><\/p>\n
      It is used to apply a function elementwise across one or more bags. In our case, the function to be called is json.dumps<\/code>. This is responsible for writing data into JSON format files. So, provide json.dumps<\/code> as input to map()<\/code> function as shown below.<\/p>\n
      my_bag.map(json.dumps).to_textfiles('data\/*.json')\n<\/code><\/pre>\n
      ['data\/0.json',\n 'data\/1.json',\n 'data\/2.json',\n 'data\/3.json',\n 'data\/4.json',\n 'data\/5.json',\n 'data\/6.json',\n 'data\/7.json',\n 'data\/8.json',\n 'data\/9.json']\n<\/code><\/pre>\n
      Yay! That was successful. Now as you might guess, dask bag is also a lazy collection. So, if you want to know or compute the actual data, you have to call the function take()<\/code> or compute()<\/code>.<\/p>\n
      For using the take()<\/code> function you need to provide input k<\/code>. This k<\/code> denotes that the first k elements should be taken<\/p>\n
      my_bag.take(3)\n<\/code><\/pre>\n
      ({'address': {'address': '812 Lakeshore Cove', 'city': 'Downers Grove'},\n  'age': 63,\n  'credit-card': {'expiration-date': '07\/25', 'number': '3749 138185 40967'},\n  'name': ('Jed', 'Munoz'),\n  'occupation': 'Clergyman',\n  'telephone': '+1-(656)-064-7533'},\n {'address': {'address': '1067 Colby Turnpike', 'city': 'Huntington Beach'},\n  'age': 62,\n  'credit-card': {'expiration-date': '01\/17', 'number': '4391 0642 7046 4592'},\n  'name': ('Emilio', 'Vega'),\n  'occupation': 'Sound Engineer',\n  'telephone': '829-959-9408'},\n {'address': {'address': '572 Boardman Route', 'city': 'Lewiston'},\n  'age': 28,\n  'credit-card': {'expiration-date': '07\/17', 'number': '4521 0738 3441 8096'},\n  'name': ('Lakia', 'Elliott'),\n  'occupation': 'Clairvoyant',\n  'telephone': '684-025-2843'})\n<\/code><\/pre>\n
      You can see that first 3 data printed in above output.<\/p>\n
      Now, let’s move on to some processing codes. For any given data, we often perform filter operations based on certain conditions. Dask bags provides the ready-made filter()<\/code> function especially for this.<\/p>\n
      Let’s say from my_bag<\/code> collection, you want to filter out the people whose age is greater than 60.
      \nFor this need to write the predicate function to check record of each age. This has to be provided as input to dask.bag.filter()<\/code> function.<\/p>\n
      my_bag.filter(lambda record: record['age'] > 60).take(4)\n<\/code><\/pre>\n
      ({'address': {'address': '812 Lakeshore Cove', 'city': 'Downers Grove'},\n  'age': 63,\n  'credit-card': {'expiration-date': '07\/25', 'number': '3749 138185 40967'},\n  'name': ('Jed', 'Munoz'),\n  'occupation': 'Clergyman',\n  'telephone': '+1-(656)-064-7533'},\n {'address': {'address': '1067 Colby Turnpike', 'city': 'Huntington Beach'},\n  'age': 62,\n  'credit-card': {'expiration-date': '01\/17', 'number': '4391 0642 7046 4592'},\n  'name': ('Emilio', 'Vega'),\n  'occupation': 'Sound Engineer',\n  'telephone': '829-959-9408'},\n {'address': {'address': '480 Rotteck Cove', 'city': 'Havelock'},\n  'age': 66,\n  'credit-card': {'expiration-date': '11\/20', 'number': '2338 5735 7231 3240'},\n  'name': ('Dewey', 'Ruiz'),\n  'occupation': 'Green Keeper',\n  'telephone': '1-445-365-1344'},\n {'address': {'address': '187 Greenwich Plaza', 'city': 'Denver'},\n  'age': 63,\n  'credit-card': {'expiration-date': '02\/20', 'number': '4879 9327 9343 8130'},\n  'name': ('Charley', 'Woods'),\n  'occupation': 'Quarry Worker',\n  'telephone': '+1-(606)-335-1595'})\n<\/code><\/pre>\n
      The earlier discussed map()<\/code> function can also be used to extract specific information. Let’s say we want to know only the occupations which people have for analysis. You can choose the occupations alone and save it in a new bag as shown below<\/p>\n
      bag_occupation=my_bag.map(lambda record: record['occupation'])\nbag_occupation.take(6)\n<\/code><\/pre>\n
      ('Clergyman',\n 'Sound Engineer',\n 'Clairvoyant',\n 'Agent',\n 'Representative',\n 'Ornamental')\n<\/code><\/pre>\n
      I have printed the first 6 data stored in the processed bag above. What if you want to know many values are there in bag_occupation<\/code>?<\/p>\n
      Your first go would be to do bag_occupation.count()<\/code>. But, remember you won’t get any result as dask.bag<\/code> is lazy. So, make sure to call compute()<\/code> at the end<\/p>\n
      # computing the no of data stored\nbag_occupation.count().compute() \n<\/code><\/pre>\n
      10000\n<\/code><\/pre>\n
      Another important function is dask.bag.groupby()<\/code><\/strong>.
      \nThis function groups collection by key function. Below is a simple example we group even and odd numbers.<\/p>\n
      !pip install partd\nb = dask.bag.from_sequence(range(10))\niseven = lambda x: x % 2 == 0\nb.groupby(iseven).compute() \n\n<\/code><\/pre>\n
      [(False, [1, 3, 5, 7, 9]), (True, [0, 2, 4, 6, 8])]\n<\/code><\/pre>\n
      It’s also possible to perform multiple data processing like filtering, mapping together in one step. This is called Chain computation. You can perform each call followed by others and finally call the compute()<\/code> function. This will save memory and time. The below code is an example of Chain Computation on the my_bag<\/code> collection.<\/p>\n
      result = (my_bag.filter(lambda record: record['age'] > 60)\n           .map(lambda record: record['occupation'])\n           .frequencies(sort=True)\n           .topk(10, key=1))\nresult.compute()\n<\/code><\/pre>\n
      [('Councillor', 6),\n ('Shop Keeper', 5),\n ('Taxi Controller', 5),\n ('Horse Riding Instructor', 4),\n ('Press Officer', 4),\n ('Nursing Manager', 4),\n ('Systems Engineer', 4),\n ('Medal Dealer', 4),\n ('Storeman', 4),\n ('Architect', 4)]\n<\/code><\/pre>\n
      Yay! we performed all processing in a single step.<\/p>\n
      Converting Dask Bag to other forms<\/h3>\n
      Many times, after processing is completed we have to convert dask bags into other forms. These other forms are generally dask dataframes, dask delayed objects, textfiles, and so on.<\/p>\n
      This section will brief you on these methods with examples.<\/p>\n
      1. How to transform Dask Bag into Dask Dataframe?<\/strong><\/p>\n
      To create Dask Dataframe from a Dask Bag, you can use **`dask.bag.to_dataframe()`** function.\n\nBag should contain tuples, dict records, or scalars. The index will not be particularly meaningful. Use reindex afterward if necessary.\n<\/code><\/pre>\n
      # Converting dask bag into dask dataframe\ndataframe=my_bag.to_dataframe()\ndataframe.compute()\n<\/code><\/pre>\n
      \"data<\/p>\n
      2. How to create Dask.Delayed<\/code> object from Dask bag<\/strong><\/p>\n
      You can convert `dask.bag` into a list of `dask.delayed` objects, one per partition using the `dask.bagto_delayed()` function. A main parameter of this function is `optimize_graph`. It is a boolean parameter. If it is set to True, then the task graph will be optimized before converting it into delayed objects.\n<\/code><\/pre>\n
      my_bag.to_delayed(True)\n<\/code><\/pre>\n
      [Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 0)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 1)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 2)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 3)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 4)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 5)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 6)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 7)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 8)),\n Delayed(('mimesis-04d0f03e80a0b650adc596eba7851142', 9))]\n<\/code><\/pre>\n
      3. How to convert Dask bag to text files<\/strong><\/p>\n
      You can write dask Bag to disk using the `dask.bag.to_textfiles()` function. As there are 10 partitions, 10 textfiles will be written. You have to provide the path or directory as input.\n<\/code><\/pre>\n
      my_bag.to_textfiles('\/content\/textfile')\n<\/code><\/pre>\n
      You have now learned how to create, operate and transform Dask bags. Next comes the most important concept in Dask.<\/p>\n
      Distributed computing with Dask – Hands on Example<\/h2>\n
      In this section, we shall load a csv file and perform the same task using pandas and Dask to compare performance. For this, first load Client<\/code> from dask.distributed<\/code>.<\/p>\n
      Dask.distributed<\/code>  will store the results of tasks in the distributed memory of the worker nodes. The central scheduler will track all the data on cluster. Once a result is completed, it is often erased from memory to create more space.<\/p>\n
      What is a Dask Client?<\/h3>\n
      The Client<\/code> is a primary entry point for users of dask.distributed<\/code>.<\/p>\n
      After we setup a cluster, we initialize a Client by pointing it to the address of a Scheduler. The Client registers itself as the default Dask scheduler, and so runs all dask collections like dask.array<\/code>, dask.bag<\/code>, dask.dataframe<\/code> and dask.delayed<\/code>.<\/p>\n
      # Import dask.distributed.Client and pandas\nfrom dask.distributed import Client\nimport pandas as pd\nimport time\n\n# Initializing a client\nclient = Client(processes=False)\nclient\n<\/code><\/pre>\n\n\n
      \n
      Client<\/h3>\n
        \n
      • Scheduler: <\/b>inproc:\/\/172.28.0.2\/1245\/1\n<\/ul>\n<\/td>\n
      		\n
      Cluster<\/h3>\n