Migrating A Large ASCII Data Set From CSV To Oracle To DSE/Cassandra – A look at storage impact

The objective of this exercise is to demonstrate how the migration of data from CSV to Oracle to Cassandra can change the required underlying storage volume of the data. How do data volumes change between pure ASCII source records versus records stored in tables in an Oracle database versus records stored in tables in a DSE/Cassandra database?

Here is the high-level plan:

  • Start out with a 1.4 GB CSV data file containing 6.1 million crime records from the Chicago Police Department for the period 2001-2016
  • Define this file as an external data file in Oracle 12c
  • Read the external data file into a physical Oracle database table
  • Use the Spark DataFrame capability introduced in Apache Spark 1.3 to load data from tables in the Oracle database via Oracle’s JDBC thin driver
  • Save the data in the dataframe to Cassandra
  • Compare the volumes used by the data in Oracle versus DSE/Cassandra

Pre-requisites

DataStax Enterprise (release 5.0.2 at time of publication)

You’ll need a working installation of DataStax Enterprise.

– Ubuntu/Debian – https://docs.datastax.com/en/datastax_enterprise/5.0/datastax_enterprise/install/installDEBdse.html
– Red Hat/Fedora/CentOS/Oracle Linux – https://docs.datastax.com/en/datastax_enterprise/5.0/datastax_enterprise/install/installRHELdse.html

To setup your environment, you’ll also need the following resources:

– Python 2.7
– Java 8
– For Red Hat, CentOS and Fedora, install EPEL (Extra Packages for Enterprise Linux).
– An Oracle database:
– In my example the database name is orcl
– A running Oracle tns listener. In my example I’m using the default port of 1521.
– You are able to make a tns connection to the Oracle database e.g. “sqlplus user/password@service_name”.
– The Oracle thin JDBC driver. You can download the ojdbc JAR file from:
http://www.oracle.com/technetwork/database/features/jdbc/jdbc-drivers-12c-download-1958347.html
I’ve used ojdbc7.jar which is certified for use with both JDK7 and JDK8
In my 12c Oracle VM Firefox this downloaded to /app/oracle/downloads/ so you’ll see the path referenced in the instructions below.

As we will connect from Spark, using the Oracle jdbc driver, to the “orcl” database on TNS port 1521, all these components must be working correctly.

Now on to installing DataStax Enterprise and playing with some data!


Set Up DataStax Components

Installation instructions for DSE are provided at the top of this doc. I’ll show the instructions for Red Hat/CentOS/Fedora here. I’m using an Oracle Enterprise Linux VirtualBox instance.

Add The DataStax Repo


As root create “`/etc/yum.repos.d/datastax.repo“`

# vi /etc/yum.repos.d/datastax.repo

Paste in these lines:

[datastax] 
name = DataStax Repo for DataStax Enterprise
baseurl=https://datastaxrepo_gmail.com:utJVKEg4lKeaWTX@rpm.datastax.com/enterprise
enabled=1
gpgcheck=0

Import The DataStax Repo Key

rpm --import http://rpm.datastax.com/rpm/repo_key 

Install DSE Components

DSE Platform

# yum install dse-full-5.0.1-1

DataStax OpsCenter

# yum install opscenter --> 6.0.2.1

DataStax OpsCenter Agent

# yum install datastax-agent --> 6.0.2.1

Enable Search & Analytics

We want to use Search (Solr) and Analytics (Spark) so we need to delete the default datacentre and restart the cluster (if its already running) in SearchAnalytics mode.

Stop the service if it’s running.

#  service dse stop
Stopping DSE daemon : dse                                  [  OK  ]

Enable Solr and Spark by changing the flag from “0” to “1” in:

# vi /etc/default/dse

e.g.:

# Start the node in DSE Search mode
SOLR_ENABLED=1

# Start the node in Spark mode
SPARK_ENABLED=1

Delete the old (Cassandra-only) datacentre databases if they exist:

# rm -rf /var/lib/cassandra/data/*
# rm -rf /var/lib/cassandra/saved_caches/*
# rm -rf /var/lib/cassandra/commitlog/*
# rm -rf /var/lib/cassandra/hints/*

Remove the old system.log if it exists:

# rm -rf /var/log/cassandra/system.log 

Now restart DSE:

$ sudo service DSE restart

After a few minutes use nodetool to check that all is up and running (check for “UN” next to the IP address):

$ nodetool status
Datacenter: SearchAnalytics
===========================
Status=Up/Down
|/ State=Normal/Leaving/Joining/Moving
--  Address    Load       Owns    Host ID                               Token                                    Rack
UN  127.0.0.1  346.89 KB  ?       8e6fa3db-9018-47f0-96df-8c78067fddaa  6840808785095143619                      rack1

Note: Non-system keyspaces don't have the same replication settings, effective ownership information is meaningless

You should also check that you can log into cqlsh:

$ cqlsh
Connected to Test Cluster at 127.0.0.1:9042.
[cqlsh 5.0.1 | Cassandra 3.0.7.1159 | DSE 5.0.1 | CQL spec 3.4.0 | Native protocol v4]
Use HELP for help.
cqlsh> 

Type exist in cqlsh to return to the shell prompt.

Identify Spark Master

We use the new DSE 5.0 format for the dse tool to get the address of the Spark Master in our cluster.
As we are using a single node for our cluster it will be no surprise that the Spark Master is also on our single node!

$ dse client-tool spark master-address
spark://127.0.0.1:7077

OK, let’s go get some data.

Load CSV Data Into Oracle

Our source file is an extract of the Chicago Crime Database 2001 to 2016, described here: http://catalog.data.gov/dataset/crimes-2001-to-present-398a4 that can be found here:https://data.cityofchicago.org/api/views/ijzp-q8t2/rows.csv?accessType=DOWNLOAD

The dataset is structured with the following fields:

– ID
– Case Number
– Date
– Block
– UCR
– Primary Type
– Description
– Location Description
– Arrest
– Domestic
– Beat
– District
– Ward
– Community Area
– FBI Code
– X Coordinate
– Y Coordinate
– Year
– Updated On
– Latitude
– Longitude
– Location

The records in the data file look like this, with a header record:

ID,Case Number,Date,Block,IUCR,Primary Type,Description,Location Description,Arrest,Domestic,Beat,District,Ward,Community Area,FBI Code,X Coordinate,Y Coordinate,Year,Updated On,Latitude,Longitude,Location

5784095,HN594666,09/17/2007 05:10:00 PM,021XX W WARREN BLVD,0486,BATTERY,DOMESTIC BATTERY SIMPLE,CHA APARTMENT,false,true,1332,012,2,28,08B,1162213,1900335,2007,04/15/2016 08:55:02 AM,41.882182894,-87.679812546,"(41.882182894, -87.679812546)"
5784096,HN594881,09/17/2007 06:35:00 PM,013XX S RACINE AVE,0460,BATTERY,SIMPLE,STREET,false,false,1231,012,2,28,08B,1168579,1894058,2007,04/15/2016 08:55:02 AM,41.864822916,-87.656618405,"(41.864822916, -87.656618405)"

We will have to build a table in Oracle for the first import of the data. When we have the data in Oracle we can assess the volume of space occupied by database tables.
We then migrate the data to Cassandra from Oracle using Spark and see how much space is occupied in Cassandra.

Wordcount tells me there are just over 6 million records (about 1.4 GB CSV file):

$ cat Crimes_-_2001_to_present.csv | wc
 6156888 69665305 1449194734

Create An Oracle User

We need an owner of the tables we’re going to create. We’ll create a user called BULK_LOAD.
To do this we need to log into Oracle SQLPlus as a privileged user (SYS or SYSTEM) to get to the SQL prompt

  • depending on your privileges you may be asked for a password when you log in as SYS or SYSTEM
  • If you’re already logged into SQLPlus as a non-system user, just type “connect / as sysdba” at the SQL> prompt
  • Log in to SQLPlus:

    $ sqlplus / AS sysdba
     
    Connected.

    Run this command to allow local User ID’s to be created:

    SQL> ALTER SESSION SET "_ORACLE_SCRIPT"=TRUE;

    Create the user BULK_LOAD and set the password to be the same as the user name:

    SQL> CREATE USER bulk_load IDENTIFIED BY bulk_load;
     
    USER created.

    Grant everything to BULK_LOAD (the old-fashioned way):

    SQL> GRANT CONNECT, resource, dba TO bulk_load;
     
    GRANT succeeded.

    Set default user and temporary tablespaces:

    SQL> ALTER USER bulk_load DEFAULT tablespace "USERS";
     
    USER altered.
     
    SQL> ALTER USER bulk_load TEMPORARY tablespace "TEMP";
     
    USER altered.

    Define An External File Directory

    This is how Oracle reads and manages flat files, allowing them to be manipulated as virtual tables. Great for loading data!

    SQL> CREATE OR REPLACE directory xtern_data_dir AS '/app/oracle/downloads';
     
    Directory created.

    Grant access to our BULK_LOAD user:

    SQL> GRANT READ,WRITE ON directory xtern_data_dir TO bulk_load;
     
    GRANT succeeded.

    Define An External File Table

    This is how Oracle reads a text file and manages it as an external table.
    Connect as our user:

    SQL> CONNECT bulk_load/bulk_load;
    Connected.

    Create the ‘virtual’ table based on the source data file.

    >The data is rather inconsistent in terms of enforcing format and type. I don’t want to have to spend a lifetime data-cleansing all 6.1 m records before starting this project so the CSV load into Oracle is in pure text format.

    In the first test we will see what happens to data when it is transferred into Oracle and then to DSE/Cassandra, all in text format. This will make an interesting baseline comparison.

    The second part of the exercise will be to build a more representative test by storing the data using a wider variety of datatypes, e.g. decimal, binary, date etc.

    DROP TABLE xternal_crime_data;
    CREATE TABLE xternal_crime_data
    (ID             			varchar2(30),
    Case_Number				varchar2(30),
    Incident_Date				varchar2(30),
    Block					varchar2(60),
    IUCR					varchar2(30),
    Primary_Type				varchar2(60),
    Description				varchar2(120),
    Location_Description		varchar2(60),
    Arrest					varchar2(60),
    Domestic				varchar2(60),
    Beat					varchar2(30),
    District				varchar2(30),
    Ward					varchar2(30),
    Community_Area				varchar2(30),
    FBI_Code				varchar2(10),
    X_Coordinate				varchar2(30),
    Y_Coordinate				varchar2(30),
    YEAR					varchar2(30),
    Updated_On				varchar2(30),
    Latitude				varchar2(30),
    Longitude				varchar2(30),
    Location				varchar2(60)
    )
    organization external
    ( DEFAULT directory xtern_data_dir
    access parameters
    ( records delimited BY newline
    FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '(' AND ')' MISSING FIELD VALUES ARE NULL)
    location ('Crimes_-_2001_to_present.csv')  
    ) REJECT LIMIT UNLIMITED;
     
    TABLE created.

    We have to do this next step:

    SQL> ALTER TABLE xternal_crime_data reject LIMIT unlimited;

    To avoid this error:

    SQL> SELECT COUNT(*) FROM xternal_crime_data;
    SELECT COUNT(*) FROM xternal_crime_data
    *
    ERROR at line 1:
    ORA-29913: error IN executing ODCIEXTTABLEFETCH callout
    ORA-30653: reject LIMIT reached

    If we describe the ‘table’ it matches the format of the file it’s based on:

    SQL> DESC xternal_crime_data
     Name					   NULL?    TYPE
     ----------------------------------------- -------- ----------------------------
     Name					   NULL?    TYPE
     ----------------------------------------- -------- ----------------------------
     ID						    VARCHAR2(30)
     CASE_NUMBER					    VARCHAR2(30)
     INCIDENT_DATE					    VARCHAR2(30)
     BLOCK						    VARCHAR2(60)
     IUCR						    VARCHAR2(30)
     PRIMARY_TYPE					    VARCHAR2(60)
     DESCRIPTION					    VARCHAR2(120)
     LOCATION_DESCRIPTION				    VARCHAR2(60)
     ARREST 					    VARCHAR2(60)
     DOMESTIC					    VARCHAR2(60)
     BEAT						    VARCHAR2(30)
     DISTRICT					    VARCHAR2(30)
     WARD						    VARCHAR2(30)
     COMMUNITY_AREA 				    VARCHAR2(30)
     FBI_CODE					    VARCHAR2(10)
     X_COORDINATE					    VARCHAR2(30)
     Y_COORDINATE					    VARCHAR2(30)
     YEAR						    VARCHAR2(30)
     UPDATED_ON					    VARCHAR2(30)
     LATITUDE					    VARCHAR2(30)
     LONGITUDE					    VARCHAR2(30)
     LOCATION					    VARCHAR2(60)

    So far Oracle hasn’t actually looked at the data yet. When we try to read the table the records will be read and validated.

    Check how many records there are in the original (source) external ‘table’ – just over 6 million:

    SQL> SELECT COUNT(*) FROM xternal_crime_data;
     
      COUNT(*)
    ----------
       6156888

    You can watch for errors or bad records while the data is being imported using tail or less on these files – log output will be to the default directory that was specified in the create external table statement:

    $ rm /app/oracle/downloads/*.bad
    $ rm /app/oracle/downloads/*.log
    

    Nothing happens until you attempt to read data from the table. You can track progress by using tail -f or less to view the log and bad files.

    Create A Table, Physically Load Data Into Oracle

    Create a physical table in the Oracle database using the “…as select” SQL syntax:

    SQL> CREATE TABLE crime_data AS SELECT * FROM xternal_crime_data;

    Let’s put a primary key and index on that table:

    SQL> CREATE INDEX crime_id ON crime_data(id);
     
    INDEX created.
    SQL> ALTER TABLE crime_data ADD CONSTRAINT crime_data_pk PRIMARY KEY(id);
     
    TABLE altered.

    We now have an index on our primary key which is now showing as a “NOT NULL” column.
    Of course in the real-world, we would make Crime ID a unique index, but for this exercise I wanted to avoid having to worry about duplicate keys.

    SQL> DESC crime_data
     Name				   NULL?    TYPE
     --------------------- -------- ----------------------------
     ID					   NOT NULL VARCHAR2(30)
     CASE_NUMBER					VARCHAR2(30)
     INCIDENT_DATE					VARCHAR2(30)
     BLOCK						    VARCHAR2(60)
     IUCR						    VARCHAR2(30)
     PRIMARY_TYPE					VARCHAR2(60)
     DESCRIPTION					VARCHAR2(120)
     LOCATION_DESCRIPTION			VARCHAR2(60)
     ARREST 					    VARCHAR2(60)
     DOMESTIC					    VARCHAR2(60)
     BEAT						    VARCHAR2(30)
     DISTRICT					    VARCHAR2(30)
     WARD						    VARCHAR2(30)
     COMMUNITY_AREA 				VARCHAR2(30)
     FBI_CODE					    VARCHAR2(10)
     X_COORDINATE					VARCHAR2(30)
     Y_COORDINATE					VARCHAR2(30)
     YEAR						    VARCHAR2(30)
     UPDATED_ON					    VARCHAR2(30)
     LATITUDE					    VARCHAR2(30)
     LONGITUDE					    VARCHAR2(30)
     LOCATION					    VARCHAR2(60)

    Check Volumes In Oracle

    At this point we have loaded the CSV source data (1.6 GB, 6.1 million rows) into an Oracle table.
    We need to find out how big is it?

    What Objects Have We Created?

    What objects does bulk_load own – the system dictionary table DBA_TABLES will tell us:

    SQL> CONNECT / AS sysdba
    SQL> SELECT TABLE_NAME FROM dba_tables WHERE owner = 'BULK_LOAD';
     
    TABLE_NAME
    --------------------------------------------------------------------------------
    CRIME_DATA
    XTERNAL_CRIME_DATA

    How Much SPace Is Oracle Using?

    Run this next query to determine total storage for a user including indexes, blobs etc. You will be prompted for the Oracle user to check, in uppercase – in my case BULK_LOAD:

    COLUMN TABLE_NAME FORMAT A32
    COLUMN OBJECT_NAME FORMAT A32
    COLUMN OWNER FORMAT A10
     
    SELECT
       owner, 
       TABLE_NAME, 
       TRUNC(SUM(bytes)/1024/1024) Meg,
       ROUND( ratio_to_report( SUM(bytes) ) OVER () * 100) Percent
    FROM
    (SELECT segment_name TABLE_NAME, owner, bytes
     FROM dba_segments
     WHERE segment_type IN ('TABLE', 'TABLE PARTITION', 'TABLE SUBPARTITION')
     UNION ALL
     SELECT i.table_name, i.owner, s.bytes
     FROM dba_indexes i, dba_segments s
     WHERE s.segment_name = i.index_name
     AND   s.owner = i.owner
     AND   s.segment_type IN ('INDEX', 'INDEX PARTITION', 'INDEX SUBPARTITION')
     UNION ALL
     SELECT l.table_name, l.owner, s.bytes
     FROM dba_lobs l, dba_segments s
     WHERE s.segment_name = l.segment_name
     AND   s.owner = l.owner
     AND   s.segment_type IN ('LOBSEGMENT', 'LOB PARTITION')
     UNION ALL
     SELECT l.table_name, l.owner, s.bytes
     FROM dba_lobs l, dba_segments s
     WHERE s.segment_name = l.index_name
     AND   s.owner = l.owner
     AND   s.segment_type = 'LOBINDEX')
    WHERE owner IN UPPER('&owner')
    GROUP BY TABLE_NAME, owner
    HAVING SUM(bytes)/1024/1024 > 10  /* Ignore really small tables */
    ORDER BY SUM(bytes) DESC
    ;
    OWNER	   TABLE_NAME				   MEG	  PERCENT
    ---------- -------------------------------- ---------- ----------
    BULK_LOAD  CRIME_DATA				  1672	      100
    

    So a 1.4GB CSV file has become 1.7GB of storage in Oracle.


    Copy Data From Oracle To Cassandra


    We now have 1.6GB of data in Oracle.
    The next objective is to see how much space that 1.7GB occupies when it is copied to Cassandra, and what impact Cassandra on-disk compression and the new Cassandra 3.0 storage engine will have on those numbers.

    So all is looking good on the Oracle side. Now time to turn to Cassandra and Spark.

    Download Oracle ojdbc7.jar

    We have to add the Oracle JDBC jar file to our Spark classpath so that Spark knows how to talk to the Oracle database.

    Using The Oracle JDBC Driver

    For this test we only need the jdbc driver file on our single (SparkMaster) node.
    In a bigger cluster we would need to distribute it to the slave nodes too.

    Update Executor Path For ojdbc7.jar In spark-defaults.conf

    Add the classpath for the ojdbc7.jar file for the executors (the path for the driver seems be required on the command line at run time as well, see below).

    # vi /etc/dse/spark/spark-defaults.conf
    

    Add the following lines pointing to the location of your ojdbc7.jar file:

    spark.driver.extraClassPath = /app/oracle/downloads/ojdbc7.jarr
    spark.executor.extraClassPath = /app/oracle/downloads/ojdbc7.jar
    

    Restart DSE

    $ sudo service dse stop
    $ sudo service dse start
    

    Start The Spark REPL

    I’m passing to the path to the ojdbc7.jar file on the command line (shouldn’t be needed as the driver path is defined in the spark-defaults.conf file now, but it seems not to work without it).

    $ dse spark --driver-class-path /app/oracle/downloads/ojdbc7.jar -deprecation
    
    scala> val crimes = sqlContext.load("jdbc", Map("url" -> "jdbc:oracle:thin:bulk_load/bulk_load@localhost:1521/orcl", "dbtable" -> "crime_data"))
    67: warning: method load in class SQLContext is deprecated: Use read.format(source).options(options).load(). This will be removed in Spark 2.0.
             val crimes = sqlContext.load("jdbc", Map("url" -> "jdbc:oracle:thin:bulk_load/bulk_load@localhost:1521/orcl", "dbtable" -> "crime_data"))

    There’s a deprecation warning message that you can ignore.
    Here’s the output we’re looking for, confirming that we established a valid connection to the remote database.

    crimes: org.apache.spark.sql.DataFrame = [ID: string, CASE_NUMBER: string, INCIDENT_DATE: string, BLOCK: string, IUCR: string, PRIMARY_TYPE: string, DESCRIPTION: string, LOCATION_DESCRIPTION: string, ARREST: string, DOMESTIC: string, BEAT: string, DISTRICT: string, WARD: string, COMMUNITY_AREA: string, FBI_CODE: string, X_COORDINATE: string, Y_COORDINATE: string, YEAR: string, UPDATED_ON: string, LATITUDE: string, LONGITUDE: string, LOCATION: string]

    Now that we’ve created a dataframe using the jdbc method shown above, we can use printSchema() to look at the dataframe schema. You’ll notice that it looks a lot like a table. That’s deliberate because it means that we can use it to manipulate large volumes of tabular data:

    scala> crimes.printSchema()
    root
     |-- ID: string (nullable = false)
     |-- CASE_NUMBER: string (nullable = true)
     |-- INCIDENT_DATE: string (nullable = true)
     |-- BLOCK: string (nullable = true)
     |-- IUCR: string (nullable = true)
     |-- PRIMARY_TYPE: string (nullable = true)
     |-- DESCRIPTION: string (nullable = true)
     |-- LOCATION_DESCRIPTION: string (nullable = true)
     |-- ARREST: string (nullable = true)
     |-- DOMESTIC: string (nullable = true)
     |-- BEAT: string (nullable = true)
     |-- DISTRICT: string (nullable = true)
     |-- WARD: string (nullable = true)
     |-- COMMUNITY_AREA: string (nullable = true)
     |-- FBI_CODE: string (nullable = true)
     |-- X_COORDINATE: string (nullable = true)
     |-- Y_COORDINATE: string (nullable = true)
     |-- YEAR: string (nullable = true)
     |-- UPDATED_ON: string (nullable = true)
     |-- LATITUDE: string (nullable = true)
     |-- LONGITUDE: string (nullable = true)
     |-- LOCATION: string (nullable = true)

    We can use the dataframe .show() method to display the first 20 rows in the table:

    scala> crimes.show()
    +--------+-----------+--------------------+--------------------+----+-------------------+--------------------+--------------------+------+---------+-----+--------+----+--------------+--------+------------+------------+-------+--------------------+------------+--------------------+--------------+
    |      ID|CASE_NUMBER|       INCIDENT_DATE|               BLOCK|IUCR|       PRIMARY_TYPE|         DESCRIPTION|LOCATION_DESCRIPTION|ARREST| DOMESTIC| BEAT|DISTRICT|WARD|COMMUNITY_AREA|FBI_CODE|X_COORDINATE|Y_COORDINATE|   YEAR|          UPDATED_ON|    LATITUDE|           LONGITUDE|      LOCATION|
    +--------+-----------+--------------------+--------------------+----+-------------------+--------------------+--------------------+------+---------+-----+--------+----+--------------+--------+------------+------------+-------+--------------------+------------+--------------------+--------------+
    |      ID|Case Number|                Date|               Block|IUCR|       Primary Type|         Description|Location Description|Arrest| Domestic| Beat|District|Ward|Community Area|FBI Code|X Coordinate|Y Coordinate|   Year|          Updated On|    Latitude|           Longitude|      Location|
    | 3666345|   HK764692|11/18/2004 12:00:...|    118XX S LOWE AVE|0890|              THEFT|       FROM BUILDING|           RESIDENCE| false|    false| 0524|     005|  34|            53|      06|     1174110|     1826325|   2004|04/15/2016 08:55:...| 41.67883435|       -87.638323571| "(41.67883435|
    | 3666346|   HK757211|11/17/2004 08:00:...|014XX N MASSASOIT...|0560|            ASSAULT|              SIMPLE|              STREET| false|    false| 2531|     025|  29|            25|     08A|     1137744|     1909068|   2004|04/15/2016 08:55:...|41.906623511|       -87.769453017|"(41.906623511|
    | 3666347|   HK764653|11/21/2004 10:00:...|  130XX S DREXEL AVE|0910|MOTOR VEHICLE THEFT|          AUTOMOBILE|CHA PARKING LOT/G...| false|    false| 0533|     005|   9|            54|      07|     1184304|     1818785|   2004|04/15/2016 08:55:...|41.657911807|       -87.601244288|"(41.657911807|
    | 3666348|   HK761590|11/19/2004 07:45:...|063XX S HAMILTON AVE|0430|            BATTERY|AGGRAVATED: OTHER...|            SIDEWALK| false|    false| 0726|     007|  15|            67|     04B|     1163120|     1862567|   2004|04/15/2016 08:55:...|41.778524374|       -87.677540732|"(41.778524374|
    | 3666350|   HK765337|11/21/2004 06:19:...| 061XX N LINCOLN AVE|1330|  CRIMINAL TRESPASS|             TO LAND|  SMALL RETAIL STORE|  true|    false| 1711|     017|  50|            13|      26|     1152804|     1941021|   2004|04/15/2016 08:55:...|41.994019837|       -87.713281946|"(41.994019837|
    |10504143|   HZ245011|04/29/2016 01:34:...|    002XX W 104TH ST|2820|      OTHER OFFENSE|    TELEPHONE THREAT|           RESIDENCE| false|    false| 0512|     005|  34|            49|      26|     1176442|     1835975|   2016|05/06/2016 03:50:...|41.705263447|       -87.629498947|"(41.705263447|
    | 3666351|   HK766200|11/21/2004 09:00:...|  014XX N STATE PKWY|0820|              THEFT|      $500 AND UNDER|              STREET| false|    false| 1824|     018|  43|             8|      06|     1176025|     1910070|   2004|04/15/2016 08:55:...| 41.90859641|       -87.628801757| "(41.90859641|
    | 3666352|   HK756872|11/17/2004 05:00:...|   045XX W WILCOX ST|1320|    CRIMINAL DAMAGE|          TO VEHICLE|              STREET| false|    false| 1113|     011|  28|            26|      14|     1146197|     1898954|   2004|04/15/2016 08:55:...|41.878712886|       -87.738658886|"(41.878712886|
    | 3666354|   HK765887|11/19/2004 10:30:...|  035XX S PAULINA ST|0610|           BURGLARY|      FORCIBLE ENTRY|             "SCHOOL|PUBLIC|BUILDING"|false|   false|0922|           009|      11|          59|          05|1165596|             1881229|        2004|04/15/2016 08:55:...|  41.829682911|
    | 3666356|   HK766293|11/21/2004 06:00:...|     006XX W 74TH ST|1310|    CRIMINAL DAMAGE|         TO PROPERTY|           APARTMENT| false|    false| 0732|     007|  17|            68|      14|     1173139|     1855820|   2004|04/15/2016 08:55:...|41.759794269|       -87.641009607|"(41.759794269|
    | 3666357|   HK766287|11/21/2004 05:00:...|  083XX S COLFAX AVE|1320|    CRIMINAL DAMAGE|          TO VEHICLE|           RESIDENCE| false|    false| 0423|     004|   7|            46|      14|     1194953|     1850154|   2004|04/15/2016 08:55:...|41.743736757|       -87.561249019|"(41.743736757|
    | 3666358|   HK763951|11/20/2004 11:05:...|082XX S DR MARTIN...|0860|              THEFT|        RETAIL THEFT|         GAS STATION| false|    false| 0631|     006|   6|            44|      06|     1180338|     1850357|   2004|04/15/2016 08:55:...|41.744641174|       -87.614792711|"(41.744641174|
    | 3666359|   HK764785|11/20/2004 09:00:...|   056XX N MOZART ST|1320|    CRIMINAL DAMAGE|          TO VEHICLE|              STREET| false|    false| 2011|     020|  40|             2|      14|     1156332|     1937421|   2004|04/15/2016 08:55:...|41.984070441|       -87.700402289|"(41.984070441|
    | 3666360|   HK755931|11/17/2004 10:00:...|  030XX N MOBILE AVE|0484|            BATTERY|PRO EMP HANDS NO/...|             "SCHOOL|PUBLIC|BUILDING"| true|   false|2511|           025|      36|          19|         08B|1133881|             1919540|        2004|04/15/2016 08:55:...|  41.935428738|
    | 3666362|   HK765213|11/21/2004 04:00:...|  046XX N ELSTON AVE|0486|            BATTERY|DOMESTIC BATTERY ...|           APARTMENT|  true|     true| 1722|     017|  39|            14|     08B|     1146545|     1930640|   2004|04/15/2016 08:55:...|41.965655711|        -87.73657143|"(41.965655711|
    | 3666363|   HK716111|10/29/2004 02:15:...|0000X N LAVERGNE AVE|2095|          NARCOTICS|ATTEMPT POSSESSIO...|            SIDEWALK|  true|    false| 1533|     015|  28|            25|      18|     1143052|     1899700|   2004|04/15/2016 08:55:...|41.880819219|       -87.750188229|"(41.880819219|
    | 3666364|   HK764786|11/21/2004 12:10:...|     005XX N RUSH ST|0810|              THEFT|           OVER $500|PARKING LOT/GARAG...| false|    false| 1834|     018|  42|             8|      06|     1177011|     1903813|   2004|04/15/2016 08:55:...|41.891404632|       -87.625369395|"(41.891404632|
    | 3666365|   HK762241|11/20/2004 03:45:...|  061XX S JUSTINE ST|0486|            BATTERY|DOMESTIC BATTERY ...|           APARTMENT| false|     true| 0713|     007|  15|            67|     08B|     1167059|     1864042|   2004|04/15/2016 08:55:...| 41.78248862|       -87.663057935| "(41.78248862|
    | 3666366|   HK765012|11/21/2004 02:20:...|016XX N SPAULDING...|0486|            BATTERY|DOMESTIC BATTERY ...|           APARTMENT| false|     true| 1422|     014|  26|            23|     08B|     1153891|     1910739|   2004|04/15/2016 08:55:...|41.910902295|       -87.710093524|"(41.910902295|
    +--------+-----------+--------------------+--------------------+----+-------------------+--------------------+--------------------+------+---------+-----+--------+----+--------------+--------+------------+------------+-------+--------------------+------------+--------------------+--------------+
    only showing top 20 rows

    Create Cassandra KeySpace

    The first thing that we need to do in Cassandra is create a keyspace to contain the table that we will create. I’m using a replication factor of 1 because I have one node in my development cluster. For most production deployments we recommend a multi-datacenter Active-Active HA setup across geographical regions using NetworkTopologyStrategy with RF=3:

    Log into cqlsh
    > If you didn’t change the IP defaults in cassandra.yaml then just type ‘cqlsh’ – if you changed the IP to be the host IP then you may need to supply the hostname e.g. ‘cqlsh [hostname]’.

    From the cqlsh prompt, create the keyspace:

    CREATE KEYSPACE IF NOT EXISTS bulk_load WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 };
    USE bulk_load;
    

    Create A Cassandra Table

    The source data had some inconsistencies that made it simpler just to load all the columns as text. To make the comparison accurate we will store the data in Cassandra in text format too.

    We will create a Cassandra table that matches the Oracle columns, with a partition key based on the crime ID.

    In cqlsh:

    cqlsh:bulk_load> CREATE TABLE crimes (
    id                      text,
    case_number             text,
    incident_date           text,
    block                   text,
    iucr                    text,
    primary_type            text,
    description             text,
    location_description    text,
    arrest                  text,
    domestic                text,
    beat                    text,
    district                text,
    ward                    text,
    community_area          text,
    fbi_code                text,
    x_coordinate            text,
    y_coordinate            text,
    YEAR                    text,
    updated_on              text,
    latitude                text,
    longitude               text, 
    location                text, 
    PRIMARY KEY (id));

    Check it’s all there:

    cqlsh:bulk_load> DESC TABLE crimes;
     
    CREATE TABLE bulk_load.crimes (
        id text PRIMARY KEY,
        arrest text,
        beat text,
        block text,
        case_number text,
        community_area text,
        description text,
        district text,
        domestic text,
        fbi_code text,
        incident_date text,
        iucr text,
        latitude text,
        location text,
        location_description text,
        longitude text,
        primary_type text,
        updated_on text,
        ward text,
        x_coordinate text,
        y_coordinate text,
        YEAR text
    ) WITH bloom_filter_fp_chance = 0.01
        AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
        AND comment = ''
        AND compaction = {'class': 'org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy', 'max_threshold': '32', 'min_threshold': '4'}
        AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
        AND crc_check_chance = 1.0
        AND dclocal_read_repair_chance = 0.1
        AND default_time_to_live = 0
        AND gc_grace_seconds = 864000
        AND max_index_interval = 2048
        AND memtable_flush_period_in_ms = 0
        AND min_index_interval = 128
        AND read_repair_chance = 0.0
        AND speculative_retry = '99PERCENTILE';

    Back to Spark…

    Change DataFrame Columns To Lower Case

    If you try to save your dataframe to Cassandra now (using the Spark-Cassandra connector) it will fail with an error message saying that columns don’t exist e.g.

    java.util.NoSuchElementException: COLUMNS NOT found IN TABLE bulk_load.crimes: ID, CASE_NUMBER, INCIDENT_DATE

    It does this even though they exist in both tables. The problem is that the connector expects the column title case in the dataframe to match the column title case in the Cassandra table. In Cassandra column names/titles are always in lower case, so the dataframe names must match.

    So we need to modify our dataframe schema…. Here’s a reminder of our raw employees dataframe schema again:
    scala> crimes.printSchema()

    scala> crimes.printSchema()
    root
     |-- ID: string (nullable = false)
     |-- CASE_NUMBER: string (nullable = true)
     |-- INCIDENT_DATE: string (nullable = true)
     |-- BLOCK: string (nullable = true)
     |-- IUCR: string (nullable = true)
     |-- PRIMARY_TYPE: string (nullable = true)
     |-- DESCRIPTION: string (nullable = true)
     |-- LOCATION_DESCRIPTION: string (nullable = true)
     |-- ARREST: string (nullable = true)
     |-- DOMESTIC: string (nullable = true)
     |-- BEAT: string (nullable = true)
     |-- DISTRICT: string (nullable = true)
     |-- WARD: string (nullable = true)
     |-- COMMUNITY_AREA: string (nullable = true)
     |-- FBI_CODE: string (nullable = true)
     |-- X_COORDINATE: string (nullable = true)
     |-- Y_COORDINATE: string (nullable = true)
     |-- YEAR: string (nullable = true)
     |-- UPDATED_ON: string (nullable = true)
     |-- LATITUDE: string (nullable = true)
     |-- LONGITUDE: string (nullable = true)
     |-- LOCATION: string (nullable = true)

    DataFrames are immutable in Spark, so we can’t modify the one we have – we need to create a new one with the correct column titles.

    Create A New Lower-Case Column Name List

    scala> val newNames = Seq("id", "case_number", "incident_date", "block","iucr","primary_type","description","location_description","arrest","domestic","beat","district","ward","community_area","fbi_code","x_coordinate","y_coordinate","year","updated_on","latitude","longitude", "location");

    Response:

    newNames: Seq[String] = List(id, case_number, incident_date, block, iucr, primary_type, description, location_description, arrest, domestic, beat, district, ward, community_area, fbi_code, x_coordinate, y_coordinate, year, updated_on, latitude, longitude, location)

    Create A New DataFrame With Lower-Case Column Names

    Now create a new dataframe based on the old one with lower case column titles:

    scala> val crimes_lc = crimes.toDF(newNames: _*)

    Response:

    crimes_lc: org.apache.spark.sql.DataFrame = [id: string, case_number: string, incident_date: string, block: string, iucr: string, primary_type: string, description: string, location_description: string, arrest: string, domestic: string, beat: string, district: string, ward: string, community_area: string, fbi_code: string, x_coordinate: string, y_coordinate: string, year: string, updated_on: string, latitude: string, longitude: string, location: string]

    And now if we look at the schema of our new datafram – hey presto! – it’s got lower case column titles.

    scala> crimes_lc.printSchema()
    root
     |-- id: string (nullable = false)
     |-- case_number: string (nullable = true)
     |-- incident_date: string (nullable = true)
     |-- block: string (nullable = true)
     |-- iucr: string (nullable = true)
     |-- primary_type: string (nullable = true)
     |-- description: string (nullable = true)
     |-- location_description: string (nullable = true)
     |-- arrest: string (nullable = true)
     |-- domestic: string (nullable = true)
     |-- beat: string (nullable = true)
     |-- district: string (nullable = true)
     |-- ward: string (nullable = true)
     |-- community_area: string (nullable = true)
     |-- fbi_code: string (nullable = true)
     |-- x_coordinate: string (nullable = true)
     |-- y_coordinate: string (nullable = true)
     |-- year: string (nullable = true)
     |-- updated_on: string (nullable = true)
     |-- latitude: string (nullable = true)
     |-- longitude: string (nullable = true)
     |-- location: string (nullable = true)

    Now we can proceed to write this data to Cassandra.

    Write The Crimes Data From Oracle To Cassandra

    Save our new dataframe CRIMES_LC to Cassandra;

    scala> crimes_lc.write.format("org.apache.spark.sql.cassandra").options(Map( "table" -> "crimes", "keyspace" -> "bulk_load")).save()
     
    scala>

    This will take a while to complete – 32 minutes in my case. While it’s in progress you may be able to check and see records appearing in the table although when it was in full swing I did get time-out errors when trying to query the table in cqlsh :[

    On my single-node VM running Oracle 12c, DSE/Cassandra and DSE/Spark I was averaging about 3000 Oracle->Spark->Cassandra rows written per second, so the import of 6.1m rows finished close to the anticipated time of 33 minutes.
    YMMV

    cqlsh:bulk_load> SELECT COUNT(*) FROM crimes;
     
     COUNT
    --------
     120428
     
    (1 ROWS)

    You can track progress from your Spark Master UI – this should be at
    http://[your-host-name]:7080/

    At the end of the run you can query the data in the Cassandra CRIMES tables by partition key:

    cqlsh:bulk_load> select * from crimes where id='3666345';
    
     id      | arrest | beat | block            | case_number | community_area | description   | district | domestic | fbi_code | incident_date          | iucr | latitude    | location      | location_description | longitude     | primary_type | updated_on             | ward | x_coordinate | y_coordinate | year
    ---------+--------+------+------------------+-------------+----------------+---------------+----------+----------+----------+------------------------+------+-------------+---------------+----------------------+---------------+--------------+------------------------+------+--------------+--------------+------
     3666345 |  false | 0524 | 118XX S LOWE AVE |    HK764692 |             53 | FROM BUILDING |      005 |    false |       06 | 11/18/2004 12:00:00 PM | 0890 | 41.67883435 | "(41.67883435 |            RESIDENCE | -87.638323571 |        THEFT | 04/15/2016 08:55:02 AM |   34 |      1174110 |      1826325 | 2004
    
    

    We can get some information about data size in Cassandra from the nodetool cfstats command:

    $ nodetool cfstats bulk_load
    Keyspace: bulk_load
    	Read Count: 1
    	Read Latency: 0.113 ms.
    	Write Count: 6156888
    	Write Latency: 0.0296818494992925 ms.
    	Pending Flushes: 0
    		Table: crimes
    		SSTable count: 6
    		Space used (live): 876227198
    		Space used (total): 876227198
    		Space used by snapshots (total): 0
    		Off heap memory used (total): 8725758
    		SSTable Compression Ratio: 0.46752276291246503
    		Number of keys (estimate): 6109619
    		Memtable cell count: 19074
    		Memtable data size: 10498992
    

    This indicates that the size of the same dataset stored in ASCII/text format in the three formats is as follows:
    – CSV (text) 1.4 GB
    – Oracle (as text) 1.7 GB
    – Cassandra (as text) 876MB

    In the case of this data, stored with a more realistic real-world Cassandra data replication factor of 3, the volume of data in a multi-node Cassandra datacenter would be 876 MB x 3 = 2,700 MB.
    Assuming the customer has fairly fast machines with fast, locally attached SSD storage, each DSE/Cassandra node might be expected to store 1 TB, so an initial cluster size of 3-5 machines would be recommended.

    Of course, it is also true that this is just one test with one table. in a “real-world” situation there may potentially be only a few tables, or a large number of tables. This data will need to be re-modelled, not just copied, when it moves from an Oracle relational model to a distributed, replicated NoSQL database model. The re-modelling, or transformation, usually involves an element of data de-normalisation and duplication, and in some cases this may have a significant impact on the volume of storage and DSE/Cassandra nodes that are reqiured to accommodate the same data in DSE/Cassandra.

    Appendix 1 – Other Stuff


    An alternative interesting-looking source file is an airline flight performance statistics
    http://www.transtats.bts.gov/DL_SelectFields.asp?Table_ID=236&DB_Short_Name=On-Time
    the file expands to 200MB

    -rw-rw-r--   1 oracle oracle  22646800 Sep 14 09:58 On_Time_On_Time_Performance_2016_1.zip
    -rw-r--r--   1 oracle oracle 200290882 Mar 14  2016 On_Time_On_Time_Performance_2016_1.csv
    

    Leave a Reply

    Your email address will not be published. Required fields are marked *

    seventeen − four =