sub = air[(air.UniqueCarrier == 'UA') & (air.Origin == 'SFO')]
byDest = sub.groupby('Dest').DepDelay.mean()
-results = byDest.compute() # this takes a while too
+results = byDest.compute() # This takes a while too.
resultsradagastshelobNote that all those machines are stand-alone login nodes. After logging in and launching a job, the job will be run on a cluster node. Usually, users cannot ssh into cluster nodes and the process of launching a job is via slurm from one of the login machines. However, if one has a job running on a cluster node, that specific node becomes accessible via ssh. This can be helpful with troubleshooting.
Dask will reads the data in parallel from the various .csv.bz2 files (unzipping on the fly), but note the caveat in the previous section about the possibilities for truly parallel I/O.
However, recall that Dask uses delayed (lazy) evaluation. In this case, the reading is delayed until compute() is called. For that matter, the various other calculations (max, groupby, mean) shown below are only done after compute() is called.
import time
t0 = time.time()
-air.DepDelay.max().compute() # this takes a while
+air.DepDelay.max().compute() # This takes a while.
print(time.time() - t0)
t0 = time.time()
-air.DepDelay.mean().compute() # this takes a while
+air.DepDelay.mean().compute() # This takes a while.
print(time.time() - t0)
air.DepDelay.median().compute() sub = air[(air.UniqueCarrier == 'UA') & (air.Origin == 'SFO')]
byDest = sub.groupby('Dest').DepDelay.mean()
-results = byDest.compute() # this takes a while too
+results = byDest.compute() # This takes a while too.
resultsYou should see this:
@@ -630,7 +634,7 @@And here is a more realistic example of filtering (subsetting).
import re
-def find(line, regex = 'Armenia'):
- vals = line.split(' ')
- if len(vals) < 6:
- return(False)
- tmp = re.search(regex, vals[3])
- if tmp is None:
- return(False)
- else:
- return(True)
-
-
-wiki.filter(find).count().compute()
-armenia = wiki.filter(find)
-smp = armenia.take(100) ## grab a handful as proof of concept
-smp[0:5]Note that it is quite inefficient to do the find() (and implicitly reading the data in) and then compute on top of that intermediate result in two separate calls to compute(). Rather, we should set up the code so that all the operations are set up before a single call to compute(). This is discussed in detail in the Dask/future tutorial.
Note that it is quite inefficient to do the find() (and implicitly read the data in) and then compute on top of that intermediate result in two separate calls to compute(). Rather, we should set up the code so that all the operations are set up before a single call to compute(). This is discussed in detail in the Dask/future tutorial.
Since the data are just treated as raw strings, we might want to introduce structure by converting each line to a tuple and then converting to a data frame.
def make_tuple(line):
@@ -674,10 +680,11 @@ Dask bags
df = armenia.map(make_tuple).to_dataframe(dtypes)
type(df)
-## Now let's actually do the computation, returning a Pandas df
-result = df.compute()
-type(result)
-result[0:5]import dask
dask.config.set(scheduler='threads', num_workers = 4)
import dask.array as da
-# x = da.from_array(x, chunks=(2500, 40000)) # adjust chunk size of existing array
+# x = da.from_array(x, chunks=(2500, 40000)) # How to adjust chunk size of existing array.
x = da.random.normal(0, 1, size=(40000,40000), chunks=(2500, 40000))
mycalc = da.mean(x, axis = 1) # row means
import time
@@ -724,18 +731,18 @@ Non-distributed arr
t0 = time.time()
x = np.random.normal(0, 1, size=(40000,40000))
time.time() - t0 # 110 sec.
-# for some reason the from_array and da.mean calculations are not done lazily here
+# For some reason the `from_array` and `da.mean` calculations are not done lazily here.
t0 = time.time()
dx = da.from_array(x, chunks=(2500, 40000))
time.time() - t0 # 27 sec.
t0 = time.time()
-mycalc = da.mean(x, axis = 1) # what is this doing given .compute() also takes time?
+mycalc = da.mean(x, axis = 1) # What is this doing given .compute() also takes time?
time.time() - t0 # 28 sec.
t0 = time.time()
rs = mycalc.compute()
time.time() - t0 # 21 sec.
Dask will avoid storing all the chunks in memory. (It appears to just generate them on the fly.) Here we have an 80 GB array but we never use more than a few GB of memory (based on top or free -h).
Dask will avoid storing all the chunks in memory. (It appears to just generate them on the fly.) Here we have an 80 GB array, but we never use more than a few GB of memory (based on top or free -h).
import dask
dask.config.set(scheduler='threads', num_workers = 4)
@@ -761,7 +768,7 @@ 3. Databases
This material is drawn from the tutorial on Working with large datasets in SQL, R, and Python, though I won’t hold you responsible for all of the database/SQL material in that tutorial, only what appears here in this Unit.
Overview
-Basically, standard SQL databases are relational databases that are a collection of rectangular format datasets (tables, also called relations), with each table similar to R or Pandas data frames, in that a table is made up of columns, which are called fields or attributes, each containing a single type (numeric, character, date, currency, enumerated (i.e., categorical), …) and rows or records containing the observations for one entity. Some of the tables in a given database will generally have fields in common so it makes sense to merge (i.e., join) information from multiple tables. E.g., you might have a database with a table of student information, a table of teacher information and a table of school information, and you might join student information with information about the teacher(s) who taught the students. Databases are set up to allow for fast querying and merging (called joins in database terminology).
+Standard SQL databases are relational databases that are a collection of rectangular format datasets (tables, also called relations), with each table similar to R or Pandas data frames, in that a table is made up of columns, which are called fields or attributes, each containing a single type (numeric, character, date, currency, enumerated (i.e., categorical), …) and rows or records containing the observations for one entity. Some of the tables in a given database will generally have fields in common so it makes sense to merge (i.e., join) information from multiple tables. E.g., you might have a database with a table of student information, a table of teacher information and a table of school information, and you might join student information with information about the teacher(s) who taught the students. Databases are set up to allow for fast querying and merging (called joins in database terminology).
Memory and disk use
Formally, databases are stored on disk, while Python and R store datasets in memory. This would suggest that databases will be slow to access their data but will be able to store more data than can be loaded into an Python or R session. However, databases can be quite fast due in part to disk caching by the operating system as well as careful implementation of good algorithms for database operations.
@@ -774,7 +781,8 @@ Interacting wi
Many DBMS have a client-server model. Clients connect to the server, with some authentication, and make requests (i.e., queries).
There are often multiple ways to interact with a DBMS, including directly using command line tools provided by the DBMS or via Python or R, among others.
We’ll concentrate on SQLite (because it is simple to use on a single machine). SQLite is quite nice in terms of being self-contained - there is no server-client model, just a single file on your hard drive that stores the database and to which you can connect to using the SQLite shell, R, Python, etc. However, it does not have some useful functionality that other DBMS have. For example, you can’t use ALTER TABLE to modify column types or drop columns.
-A good alternative to SQLite that I encourage you to consider is DuckDB. DuckDB stores data column-wise, which can lead to big speedups when doing queries operating on large portions of tables (so-called “online analytical processing” (OLAP)). Another nice feature of DuckDB is that it can interact with data on disk without always having to read all the data into memory. In fact, ideally we’d use it for this class, but I haven’t had time to create a DuckDB version of the StackOverflow database.
+A good alternative to SQLite that I encourage you to consider is DuckDB. DuckDB stores data column-wise, which can lead to big speedups when doing queries operating on large portions of tables (so-called “online analytical processing” (OLAP)). Another nice feature of DuckDB is that it can interact with data on disk without always having to read all the data into memory.
+In the demo code, we’ll have the option to use either SQLite or DuckDB.
Database schema and normalization
@@ -940,7 +948,7 @@ Stack Over
You can view one reasonable schema. The lines between tables indicate the relationship of foreign keys in one table to primary keys in another table. The schema in the actual database of Stack Overflow data we’ll use in the examples here is similar to but not identical to that.
-You can download a copy of the SQLite version of the Stack Overflow 2021 database.
+You can download a copy of the SQLite version of the Stack Overflow 2021 database or a copy of the DuckDB version of the Stack Overflow 2021 database.
import sqlite3 as sq
dir_path = '../data' # Replace with the actual path
db_filename = 'stackoverflow-2021.db'
-## download from http://www.stat.berkeley.edu/share/paciorek/stackoverflow-2021.db
+## Download from http://www.stat.berkeley.edu/share/paciorek/stackoverflow-2021.db.
con = sq.connect(os.path.join(dir_path, db_filename))
db = con.cursor()
db.execute("select * from questions limit 5") # simple query <sqlite3.Cursor object at 0x7f8202e37640><sqlite3.Cursor object at 0x7f1bc801f640>db.fetchall() # retrieve results[(65534165.0, '2021-01-01 22:15:54', 0.0, 112.0, 2.0, 0.0, None, "Can't update a value in sqlite3", 13189393.0), (65535296.0, '2021-01-02 01:33:13', 2.0, 1109.0, 0.0, 0.0, None, 'Install and run ROS on Google Colab', 14924336.0), (65535910.0, '2021-01-02 04:01:34', -1.0, 110.0, 1.0, 8.0, 0.0, 'Operators on date/time fields', 651174.0), (65535916.0, '2021-01-02 04:03:20', 1.0, 35.0, 1.0, 0.0, None, 'Plotting values normalised', 14695007.0), (65536749.0, '2021-01-02 07:03:04', 0.0, 108.0, 1.0, 5.0, None, 'Export C# to word with template', 14899717.0)]Alternatively, we could use DuckDB. However, I don’t have a DuckDB version of the StackOverflow database, so one can’t actually run this code.
+Alternatively, we could use DuckDB.
import duckdb as dd
dir_path = '../data' # Replace with the actual path
-db_filename = 'stackoverflow-2021.duckdb' # This doesn't exist.
-
-con = dd.connect(os.path.join(dir_path, db_filename))
-db = con.cursor()
-db.execute("select * from questions limit 5") # simple query
-db.fetchall() # retrieve resultsWe can (fairly) easily see the tables (this is easier from R):
One can create tables from within the ‘sqlite’ command line interfaces (discussed in the tutorial), but often one would do this from Python or R. Here’s the syntax from Python, creating the table from a Pandas dataframe.
## create data frame 'student_data' in some fashion
+## Create data frame 'student_data' in some fashion.
con = sq.connect(db_path)
student_data.to_sql('student', con, if_exists='replace', index=False)