<jupyter_start><jupyter_text>10 Minutes to cuDF and Dask-cuDF=======================Modeled after 10 Minutes to Pandas, this is a short introduction to cuDF and Dask-cuDF, geared mainly for new users. What are these Libraries?[cuDF](https://github.com/rapidsai/cudf) is a Python GPU DataFrame library (built on the Apache Arrow columnar memory format) for loading, joining, aggregating, filtering, and otherwise manipulating data.[Dask](https://dask.org/) is a flexible library for parallel computing in Python that makes scaling out your workflow smooth and simple.[Dask-cuDF](https://github.com/rapidsai/dask-cudf) is a library that provides a partitioned, GPU-backed dataframe, using Dask. When to use cuDF and Dask-cuDFIf your workflow is fast enough on a single GPU or your data comfortably fits in memory on a single GPU, you would want to use cuDF. If you want to distribute your workflow across multiple GPUs, have more data than you can fit in memory on a single GPU, or want to analyze data spread across many files at once, you would want to use Dask-cuDF.<jupyter_code>import os

import numpy as np
import pandas as pd
import cudf
import dask_cudf

np.random.seed(12)

#### Portions of this were borrowed and adapted from the
#### cuDF cheatsheet, existing cuDF documentation,
#### and 10 Minutes to Pandas.<jupyter_output><empty_output><jupyter_text>Object Creation--------------- Creating a `cudf.Series` and `dask_cudf.Series`.<jupyter_code>s = cudf.Series([1,2,3,None,4])
print(s)
ds = dask_cudf.from_cudf(s, npartitions=2) 
print(ds.compute())<jupyter_output><empty_output><jupyter_text>Creating a `cudf.DataFrame` and a `dask_cudf.DataFrame` by specifying values for each column.<jupyter_code>df = cudf.DataFrame({'a': list(range(20)),
'b': list(reversed(range(20))),
'c': list(range(20))})
df
ddf = dask_cudf.from_cudf(df, npartitions=2) 
print(ddf.compute())<jupyter_output><empty_output><jupyter_text>Creating a `cudf.DataFrame` from a pandas `Dataframe` and a `dask_cudf.Dataframe` from a `cudf.Dataframe`.*Note that best practice for using Dask-cuDF is to read data directly into a `dask_cudf.DataFrame` with something like `read_csv` (discussed below).*<jupyter_code>pdf = pd.DataFrame({'a': [0, 1, 2, 3],'b': [0.1, 0.2, None, 0.3]})
gdf = cudf.DataFrame.from_pandas(pdf)
print(gdf)
dask_df = dask_cudf.from_cudf(pdf, npartitions=2)
dask_gdf = dask_cudf.from_dask_dataframe(dask_df)
print(dask_gdf.compute())<jupyter_output><empty_output><jupyter_text>Viewing Data------------- Viewing the top rows of a GPU dataframe.<jupyter_code>print(df.head(2))
print(ddf.head(2))<jupyter_output><empty_output><jupyter_text>Sorting by values.<jupyter_code>print(df.sort_values(by='b'))
print(ddf.sort_values(by='b').compute())<jupyter_output><empty_output><jupyter_text>Selection------------ Getting Selecting a single column, which initially yields a `cudf.Series` or `dask_cudf.Series`. Calling `compute` results in a `cudf.Series` (equivalent to `df.a`).<jupyter_code>print(df['a'])
print(ddf['a'].compute())<jupyter_output><empty_output><jupyter_text>Selection by Label Selecting rows from index 2 to index 5 from columns 'a' and 'b'.<jupyter_code>print(df.loc[2:5, ['a', 'b']])
print(ddf.loc[2:5, ['a', 'b']].compute())<jupyter_output><empty_output><jupyter_text>Selection by Position Selecting via integers and integer slices, like numpy/pandas. Note that this functionality is not available for Dask-cuDF DataFrames.<jupyter_code>print(df.iloc[0])
print(df.iloc[0:3, 0:2])<jupyter_output><empty_output><jupyter_text>You can also select elements of a `DataFrame` or `Series` with direct index access.<jupyter_code>print(df[3:5])
print(s[3:5])<jupyter_output><empty_output><jupyter_text>Boolean Indexing Selecting rows in a `DataFrame` or `Series` by direct Boolean indexing.<jupyter_code>print(df[df.b > 15])
print(ddf[ddf.b > 15].compute())<jupyter_output><empty_output><jupyter_text>Selecting values from a `DataFrame` where a Boolean condition is met, via the `query` API.<jupyter_code>print(df.query("b == 3"))  
print(ddf.query("b == 3").compute())<jupyter_output><empty_output><jupyter_text>You can also pass local variables to Dask-cuDF queries, via the `local_dict` keyword. With standard cuDF, you may either use the `local_dict` keyword or directly pass the variable via the `@` keyword.<jupyter_code>cudf_comparator = 3
print(df.query("b == @cudf_comparator"))
dask_cudf_comparator = 3
print(ddf.query("b == @val", local_dict={'val':dask_cudf_comparator}).compute())<jupyter_output><empty_output><jupyter_text>Supported logical operators include `>`, `=`, `<=`, `==`, and `!=`.  MultiIndex cuDF supports hierarchical indexing of DataFrames using MultiIndex. Grouping hierarchically (see `Grouping` below) automatically produces a DataFrame with a MultiIndex.<jupyter_code>arrays = [['a', 'a', 'b', 'b'],
          [1, 2, 3, 4]]
tuples = list(zip(*arrays))
idx = cudf.MultiIndex.from_tuples(tuples)
idx<jupyter_output><empty_output><jupyter_text>This index can back either axis of a DataFrame.<jupyter_code>gdf1 = cudf.DataFrame({'first': np.random.rand(4), 'second': np.random.rand(4)})
gdf1.index = idx
print(gdf1.to_pandas())
gdf2 = cudf.DataFrame({'first': np.random.rand(4), 'second': np.random.rand(4)}).T
gdf2.columns = idx
print(gdf2.to_pandas())<jupyter_output><empty_output><jupyter_text>Accessing values of a DataFrame with a MultiIndex. Note that slicing is not yet supported.<jupyter_code>print(gdf1.loc[('b', 3)].to_pandas())<jupyter_output><empty_output><jupyter_text>Missing Data------------ Missing data can be replaced by using the `fillna` method.<jupyter_code>print(s.fillna(999))
print(ds.fillna(999).compute())<jupyter_output><empty_output><jupyter_text>Operations------------  Stats Calculating descriptive statistics for a `Series`.<jupyter_code>print(s.mean(), s.var())
print(ds.mean().compute(), ds.var().compute())<jupyter_output><empty_output><jupyter_text>Applymap Applying functions to a `Series`. Note that applying user defined functions directly with Dask-cuDF is not yet implemented. For now, you can use [map_partitions](http://docs.dask.org/en/stable/dataframe-api.htmldask.dataframe.DataFrame.map_partitions) to apply a function to each partition of the distributed dataframe.<jupyter_code>def add_ten(num):
    return num + 10

print(df['a'].applymap(add_ten))
print(ddf['a'].map_partitions(add_ten).compute())<jupyter_output><empty_output><jupyter_text>Histogramming Counting the number of occurrences of each unique value of variable.<jupyter_code>print(df.a.value_counts())
print(ddf.a.value_counts().compute())<jupyter_output><empty_output><jupyter_text>String Methods Like pandas, cuDF provides string processing methods in the `str` attribute of `Series`. Full documentation of string methods is a work in progress. Please see the cuDF API documentation for more information.<jupyter_code>s = cudf.Series(['A', 'B', 'C', 'Aaba', 'Baca', None, 'CABA', 'dog', 'cat'])
print(s.str.lower())
ds = dask_cudf.from_cudf(s, npartitions=2)
print(ds.str.lower().compute())<jupyter_output><empty_output><jupyter_text>Concat Concatenating `Series` and `DataFrames` row-wise.<jupyter_code>s = cudf.Series([1, 2, 3, None, 5])
print(cudf.concat([s, s]))
ds2 = dask_cudf.from_cudf(s, npartitions=2)
print(dask_cudf.concat([ds2, ds2]).compute())<jupyter_output><empty_output><jupyter_text>Join Performing SQL style merges. Note that the dataframe order is not maintained, but may be restored post-merge by sorting by the index.<jupyter_code>df_a = cudf.DataFrame()
df_a['key'] = ['a', 'b', 'c', 'd', 'e']
df_a['vals_a'] = [float(i + 10) for i in range(5)]

df_b = cudf.DataFrame()
df_b['key'] = ['a', 'c', 'e']
df_b['vals_b'] = [float(i+100) for i in range(3)]

merged = df_a.merge(df_b, on=['key'], how='left')
print(merged)
ddf_a = dask_cudf.from_cudf(df_a, npartitions=2)
ddf_b = dask_cudf.from_cudf(df_b, npartitions=2)

merged = ddf_a.merge(ddf_b, on=['key'], how='left').compute()
print(merged)<jupyter_output><empty_output><jupyter_text>Append Appending values from another `Series` or array-like object.<jupyter_code>print(s.append(s))
print(ds2.append(ds2).compute())<jupyter_output><empty_output><jupyter_text>Grouping Like pandas, cuDF and Dask-cuDF support the Split-Apply-Combine groupby paradigm.<jupyter_code>df['agg_col1'] = [1 if x % 2 == 0 else 0 for x in range(len(df))]
df['agg_col2'] = [1 if x % 3 == 0 else 0 for x in range(len(df))]

ddf = dask_cudf.from_cudf(df, npartitions=2)<jupyter_output><empty_output><jupyter_text>Grouping and then applying the `sum` function to the grouped data.<jupyter_code>print(df.groupby('agg_col1').sum())
print(ddf.groupby('agg_col1').sum().compute())<jupyter_output><empty_output><jupyter_text>Grouping hierarchically then applying the `sum` function to grouped data. We send the result to a pandas dataframe only for printing purposes.<jupyter_code>print(df.groupby(['agg_col1', 'agg_col2']).sum().to_pandas())
ddf.groupby(['agg_col1', 'agg_col2']).sum().compute().to_pandas()<jupyter_output><empty_output><jupyter_text>Grouping and applying statistical functions to specific columns, using `agg`.<jupyter_code>print(df.groupby('agg_col1').agg({'a':'max', 'b':'mean', 'c':'sum'}))
print(ddf.groupby('agg_col1').agg({'a':'max', 'b':'mean', 'c':'sum'}).compute())<jupyter_output><empty_output><jupyter_text>Transpose Transposing a dataframe, using either the `transpose` method or `T` property. Currently, all columns must have the same type. Transposing is not currently implemented in Dask-cuDF.<jupyter_code>sample = cudf.DataFrame({'a':[1,2,3], 'b':[4,5,6]})
print(sample)
print(sample.transpose())<jupyter_output><empty_output><jupyter_text>Time Series------------ `DataFrames` supports `datetime` typed columns, which allow users to interact with and filter data based on specific timestamps.<jupyter_code>import datetime as dt

date_df = cudf.DataFrame()
date_df['date'] = pd.date_range('11/20/2018', periods=72, freq='D')
date_df['value'] = np.random.sample(len(date_df))

search_date = dt.datetime.strptime('2018-11-23', '%Y-%m-%d')
print(date_df.query('date <= @search_date'))
date_ddf = dask_cudf.from_cudf(date_df, npartitions=2)
print(date_ddf.query('date <= @search_date', local_dict={'search_date':search_date}).compute())<jupyter_output><empty_output><jupyter_text>Categoricals------------ `DataFrames` support categorical columns.<jupyter_code>pdf = pd.DataFrame({"id":[1,2,3,4,5,6], "grade":['a', 'b', 'b', 'a', 'a', 'e']})
pdf["grade"] = pdf["grade"].astype("category")

gdf = cudf.DataFrame.from_pandas(pdf)
print(gdf)
dgdf = dask_cudf.from_cudf(gdf, npartitions=2)
print(dgdf.compute())<jupyter_output><empty_output><jupyter_text>Accessing the categories of a column. Note that this is currently not supported in Dask-cuDF.<jupyter_code>gdf.grade.cat.categories<jupyter_output><empty_output><jupyter_text>Accessing the underlying code values of each categorical observation.<jupyter_code>print(gdf.grade.cat.codes)
print(dgdf.grade.cat.codes.compute())<jupyter_output><empty_output><jupyter_text>Converting Data Representation--------------------------------  Pandas Converting a cuDF and Dask-cuDF `DataFrame` to a pandas `DataFrame`.<jupyter_code>print(df.head().to_pandas())
print(ddf.compute().head().to_pandas())<jupyter_output><empty_output><jupyter_text>Numpy Converting a cuDF or Dask-cuDF `DataFrame` to a numpy `ndarray`.<jupyter_code>print(df.as_matrix())
print(ddf.compute().as_matrix())<jupyter_output><empty_output><jupyter_text>Converting a cuDF or Dask-cuDF `Series` to a numpy `ndarray`.<jupyter_code>print(df['a'].to_array())
print(ddf['a'].compute().to_array())<jupyter_output><empty_output><jupyter_text>Arrow Converting a cuDF or Dask-cuDF `DataFrame` to a PyArrow `Table`.<jupyter_code>print(df.to_arrow())
print(ddf.compute().to_arrow())<jupyter_output><empty_output><jupyter_text>Getting Data In/Out------------------------  CSV Writing to a CSV file, by first sending data to a pandas `Dataframe` on the host.<jupyter_code>if not os.path.exists('example_output'):
    os.mkdir('example_output')
    
df.to_pandas().to_csv('example_output/foo.csv', index=False)
ddf.compute().to_pandas().to_csv('example_output/foo_dask.csv', index=False)<jupyter_output><empty_output><jupyter_text>Reading from a csv file.<jupyter_code>df = cudf.read_csv('example_output/foo.csv')
print(df)
ddf = dask_cudf.read_csv('example_output/foo_dask.csv')
print(ddf.compute())<jupyter_output><empty_output><jupyter_text>Reading all CSV files in a directory into a single `dask_cudf.DataFrame`, using the star wildcard.<jupyter_code>ddf = dask_cudf.read_csv('example_output/*.csv')
print(ddf.compute())<jupyter_output><empty_output><jupyter_text>Parquet Writing to parquet files, using the CPU via PyArrow.<jupyter_code>df.to_parquet('example_output/temp_parquet')<jupyter_output><empty_output><jupyter_text>Reading parquet files with a GPU-accelerated parquet reader.<jupyter_code>df = cudf.read_parquet('example_output/temp_parquet')
print(df.to_pandas())<jupyter_output><empty_output><jupyter_text>Writing to parquet files from a `dask_cudf.DataFrame` using PyArrow under the hood.<jupyter_code>ddf.to_parquet('example_files')<jupyter_output><empty_output><jupyter_text>ORC Reading ORC files.<jupyter_code>!wget https://github.com/rapidsai/cudf/raw/branch-0.14/python/cudf/cudf/tests/data/orc/TestOrcFile.test1.orc
df2 = cudf.read_orc('TestOrcFile.test1.orc')
df2.to_pandas()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1552429857855176, 'fraction_numerical': 0.011758577063332588, 'mean_word_length': 7.129461584996975, 'pattern_counts': {'":': 2, '<': 180, '<?xml version=': 0, '>': 180, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '44', 'dir': 'jupyter-structured-clean-dedup', 'id': '74699', 'license': 'Apache-2.0', 'n_tokens_mistral': 5021, 'n_tokens_neox': 4870, 'n_words': 1051, 'path': 'Jetson_Anaconda_Dockerfiles/10min_dask_cudf.ipynb', 'repo_name': 'RyanAtNVIDIA/rapids_l4t_feedstock', 'size': '13437'}