How to make data available for analytics ASAP

How to make data available for analytics
ASAP
Jens Röwekamp
Software Engineer
MariaDB Corporation
Markus Mäkelä
Senior Software Engineer
MariaDB Corporation

1. MariaDB ColumnStore
○ What is ColumnStore
2. Loading data into ColumnStore
○ Command Line Tools & SQL
○ Bulk Write API & Native Wrapper APIs
○ Bulk Write API and Apache Spark
○ Application integration via Data Adapters
3. Future Improvements
Quick Overview
2

What is ColumnStore?
● Columnar storage
○ Efficient use of large and wide tables
■ SELECT column_1 FROM 2000_column_table
○ Columns split into 8M row blocks (extents)
● Massively parallel
○ Columns and extents read in parallel
○ Push work to worker nodes (PMs)
■ Predicate filtering
■ join processing
■ initial aggregation of data
● Designed for analytical workloads
4

● PM (Performance Module) nodes
○ Handles primitive jobs
○ Preliminary aggregation
○ Predicate filtering
● UM (User Module) nodes
○ Handle SQL statements
○ Combines primitive results
● Storage Layer
○ Local
○ SAN
○ GlusterFS
What is ColumnStore?
5

6
MariaDB ColumnStore Architecture
Columnar Distributed Data Storage
Local Storage | SAN | NAS | EBS | Gluster FS
BI Tool SQL Client Custom
Big Data App
Application
MariaDB SQL
Front End
Distributed
Query Engine
Data Storage

Loading data into
ColumnStore
Command Line Tools & SQL
7

cpimport
● Command line client
● CSV input
● In all versions
● Local data loading
8
[root@cs ~]# cpimport test tmp1 << EOF
1|a
2|b
3|c
EOF

cpimport: Mode 1
● Data and cpimport on one UM/PM
9
Source
File
cpimport UM Node
PM
Node
PM
Node
PM
Node
...

cpimport: Mode 2
● cpimport on one UM/PM
● Partitioned data on PM
10
cpimport UM Node
PM
Node
PM
Node
PM
Node
...
Partitioned
Source File
Partitioned
Source File
Partitioned
Source File

cpimport: Mode 3
● Partitioned data and cpimport on PM
11
UM Node
PM
Node
PM
Node
PM
Node
...
Partitioned
Source File
Partitioned
Source File
Partitioned
Source File
cpimport cpimport

LOAD DATA INFILE
● Familiar SQL interface
● With autocommit on, it uses cpimport
● With autocommit off, it uses native DML
● LOAD DATA LOCAL INFILE uses native DML
12
MariaDB [test]> LOAD DATA INFILE
‘import.tbl’ INTO TABLE tmp1 FIELDS
TERMINATED BY ‘|’;

mcsimport
● Command line client
● CSV input
● Since ColumnStore 1.2
● Remote data loading
13
PS C:> mcsimport.exe test tmp1
C:data-to-importtable.csv -c
C:cs-configsColumnstore.xml

Loading data into
ColumnStore
Bulk Write API & Native Wrapper APIs
14

Language Bindings
● Simple C++ (11) API to bulk load data directly into the PMs
● Currently available on modern Linux distributions and Windows
● Other language bindings are implemented using SWIG which generates
efficient almost identical native implementations on top of the C++ library:
○ Java 8 (also providing Scala support)
○ Python 2 & 3
○ Other language bindings can be implemented in the future
15

System Configuration
● The API assumes the existence of a Columnstore.xml file in the system in order
to determine the system topology, hosts, and ports for the PM nodes.
● If you are running on a ColumnStore node, the adapter will work immediately
● For a remote host, you will need to copy the Columnstore.xml from a server
node.
● The API will need to be able to connect with the ProcMon (8800), WriteEngine
(8630), and DBRMController (8616) ports.
16

Core Classes
The following classes provide the core interface:
● ColumnStoreDriver: Entry point / connection
● ColumnStoreBulkInsert: Per table interface for writing a transaction
● ColumnStoreSystemCatalog: Table metadata retrieval
Language namespaces:
● C++ - mcsapi::
● Java - com.mariadb.columnstore.api
● Python - pymcsapi
17

Core Classes - ColumnStoreDriver
● Entry point and factory class for creating:
○ ColumnStoreBulkInsert objects to allow bulk write of a single transaction for a single
table
○ ColumnStoreSystemCatalog object to allow retrieval of table and column data
● Default constructor will look for Columnstore.xml in:
○ /usr/local/mariadb/columnstore/etc/Columnstore.xml
○ $COLUMNSTORE_INSTALL_DIR/etc/Columnstore.xml
● Alternatively pass path to Columnstore.xml as constructor argument to specify
non standard location
● Also able to list and clear ColumnStore table locks since version 1.2.2.
18

Core Classes - ColumnStoreBulkInsert
● Encapsulates bulk insert operations. Constructed for a single table and
transaction.
● Multiple instances can be created for multiple drivers but you can only have one
active per table per ColumnStore instance.
● Error handling is important. If you fail to commit or rollback, a ColumnStore
table lock will be left. This can be released manually with the cleartablelock
command or through ColumnStoreDriver.
● After completion getSummary returns details about the injection.
19

Core Classes - ColumnStoreSystemCatalog
● Allows retrieval of ColumnStore table and column metadata to allow for generic
manipulations.
20
ColumnStoreSystemCatalog.getTable(db, table)
ColumnStoreSystemCatalogTable.getColumnCount()
…
ColumnStoreSystemCatalogTable.getColumn(id | name)
ColumnStoreSystemCatalogTableColumn.getType()
…
ColumnStoreSystemCatalogTableColumn.getDefaultValue()

pymcsapi example
import pymcsapi
try:
driver = pymcsapi.ColumnStoreDriver()
bulk = driver.createBulkInsert("test", "t1", 0, 0)
for i in range(0,1000):
bulk.setColumn(0, i)
bulk.setColumn(1, 1000-i)
bulk.writeRow()
bulk.commit()
except RuntimeError as err:
bulk.rollback()
print("Error caught: %s" % (err,))
21

Loading data into
ColumnStore
Bulk Write API and Apache Spark
22

Apache Spark & MariaDB ColumnStore
● Enables best of breed approach:
○ In memory machine learning algorithms of Spark
○ Publish results to ColumnStore for ease of consumption with SQL tools such as
Tableau
● To read data from ColumnStore into Spark
○ JDBC connector
○ Spark SQL
● To write data from Spark into ColumnStore
○ Bulk Write API
○ ColumnStoreExporter
23

Export via ColumnStoreExporter
● Object on top of the Bulk Write API to export data with minimal lines of code.
● Requires the same structure of DataFrame/RDD and ColumnStore table.
● Methods:
○ generateTableStatement(DataFrame, [database], [table], [determineTypeLength])
○ export(database, table, DataFrame, [path to Columnstore.xml])
○ exportFromWorkers(database, table, RDD, [partitions], [path to Columnstore.xml])*
*only available in Scala
24

ColumnStoreExporter example
import columnStoreExporter
from pyspark.sql import SparkSession, Row
import mysql.connector as mariadb
spark = SparkSession.builder.appName("DataFrame export into ColumnStore").getOrCreate()
df = spark.createDataFrame(
spark.sparkContext.parallelize(range(0, 128)).map(lambda i: Row(number=i, ASCII=chr(i)))
)
25

ColumnStoreExporter example
try:
conn = mariadb.connect(user='root', database='', host='127.0.0.1', password='')
cursor = conn.cursor()
cursor.execute(columnStoreExporter.generateTableStatement(df, "test", "pyspark_export"))
except mariadb.Error as err:
print("Error during table creation: ", err)
finally:
if cursor: cursor.close()
if conn: conn.close()
columnStoreExporter.export("test","pyspark_export",df)
spark.stop()
26

Loading data into
ColumnStore
Application integration via Data Adapters
27

Pentaho Data Integration Adapter
● PlugIn for PDI and Pentaho Server.
● Allows the export of data from PDI transformations into ColumnStore.
● Dedicated session tomorrow at 10:10am.
28

MaxScale CDC Adapter
● Connects to MaxScale CDC service
○ MariaDB binary logs to ColumnStore data
● Replicate mode
○ Convenient
■ INSERTs via Bulk Insert API
■ UPDATE and DELETE via SQL interface
○ Intended for INSERT heavy workloads
● Transformation mode
○ Fastest
■ INSERT →{"type": "insert", ...}
■ DELETE → {"type": "delete, ...}
■ UPDATE → {"type": "update_before", ...} + {"type": "update_after", ...}
32

Replicate Mode
33
UM
PM
PM
PM
MaxScale

Transform Mode
34
UM
PM
PM
PM
MaxScale
f(x)

● Java connectivity via Kafka
○ Reads Confluent Avro serialized data from Kafka
○ Inserts into ColumnStore
○ Java → Kafka → ColumnStore
● One topic per table
○ Topics applied in parallel
Kafka to ColumnStore Adapter
35

Kafka to ColumnStore Adapter
36

Integrated MaxScale CDC
38
● Live transformation
○ No storage required
○ Faster
● Transform & Replicate
● Planned for MaxScale 2.4

Integrated MaxScale CDC
39
UM
PM
PM
PM
MaxScale
f(x)

Performance enhancements
40
● Bulk Write API
○ Async loading
● mcsimport
○ Code and compiler optimizations
○ Pipelining

Ways to inject data into ColumnStore
● Command line tools & SQL
○ cpimport
○ mcsimport
○ LOAD DATA INFILE
● Bulk Write API
○ C++, Java and Python
○ ColumnStoreExporter for Apache Spark
● Data Adapters
○ Pentaho Data Integration
○ MaxScale CDC Adapter
○ Kafka to ColumnStore Adapter
42

Further cpimport examples
46
mcsmysql -q -e 'select * from <source-table>;' -N <source-db> |
cpimport -s 't' <target-db> <target-table>
Import an InnoDB table into ColumnStore
aws s3 cp --quiet s3://dthompson-test/trades_bulk.csv - |
cpimport test trades -s ","
Import data from an AWS S3 bucket

Motivation
● Organizations need to make data available for analysis as soon as it arrives.
● Enable Machine learning results to be published and accessible by business
users through SQL based tools.
● Ease of integration whether custom or ETL tools.
47

Bulk Write API
Applications can use the bulk
write API to collect and write
data - on-demand data
loading
No need to copy CSV to
ColumnStore node -
simpler
Bypass SQL interface,
parser and optimizer -
faster writes
MariaDB Server
ColumnStore UM
Application
ColumnStore PM ColumnStore PMColumnStore PM
Write API Write API Write API
MariaDB Server
ColumnStore UM
Bulk Data Adapter
1. For each row
a. For each column
bulkInsert->setColumn
b. bulkInsert->writeRow
2. bulkInsert->commit
* Buffer 100,000 rows by default
48

MARIADB
PRESENTATION
PRESENTER NAME
Presenter Title
MariaDB Corporation
49

MARIADB
PRESENTATION
PRESENTER NAME
Presenter Title
MariaDB Corporation
50

MARIADB
PRESENTATION
PRESENTER NAME
Presenter Title
MariaDB Corporation
51

PURPOSE-BUILT STORAGE: MYROCKS
● SSD optimized: space, writes and lifetime
● Writes: trades random IO on writes for random IO on reads
● Storage: does not use a fixed page size (InnoDB is sector aligned: 4KB)
● Storage: has smaller metadata for primary key indexes
○ InnoDB: 13 bytes, and not compressed
○ MyRocks: 8 bytes + zero filling + prefix key encoding, then compressed
54

SCHEMA EVOLUTION:
INVISIBLE COLUMNS
CREATE TABLE users(
id INT PRIMARY KEY,
name VARCHAR(50),
bio TEXT(2000) COMPRESSED,
secret VARCHAR(10) INVISIBLE);
SQL Server = HIDDEN (period columns only), DB2 = IMPLICITLY HIDDEN, ORACLE = INVISIBLE
56

TEMPORAL TABLES
AND QUERIES
57

THIS IS A TITLE
Lorem ipsum dolor sit amet, consectetur adipiscing
elit. Nam pretium augue nunc, quis bibendum ligula
molestie sit amet. Nam sed luctus tellus.
Praesent nec cursus ex, vel commodo tellus. Duis
tempus pharetra ante a ullamcorper. Curabitur
commodo purus eget tempus faucibus.
58

PURPOSE-BUILT STORAGE: WHY?
Purpose-built database
Relational
Database
(mixed)
Wide-column
Database
(write-intensive)
Document
Database
(scalable)
Columnar
Database
(analytical)
59

We are running more than 25 billion
queries an hour on MariaDB...the query
patterns change every hour.
—Tim Yim, Director of Operations, ServiceNow
60

Blue Azure
#0E6488
Sea Fresh
#96DDCF
Granite
#424F62
Brand Guidelines
Colors and Type
Open Seas
#2F99A3
Deep Ocean
#003545
Electric Eel
#AC74A
66

Icons
Company
MariaDB Server ColumnStore MaxScale
69

Icons
Miscellaneous (Circle)
70

Icons
Miscellaneous (Filled)
71

Icons
Social
Facebook Twitter LinkedIn YouTubeGoogle+
72

How to make data available for analytics ASAP

Recommended

More Related Content

What's hot (20)

Similar to How to make data available for analytics ASAP (20)

More from MariaDB plc (20)

Recently uploaded (20)

How to make data available for analytics ASAP