Dremio Data Lake Engine Apache Arrow Flight Connector with Spark Machine Learning
There are following components in the writing:
Apache Arrow
Apache Arrow Flight
Dremio server
Dremio Flight Connector
Apache Spark Machine Learning.
Let’s itemize all components in the writing:

Apache Arrow:

Apache Arrow is a cross-language development platform for in-memory data. It specifies a standardized language-independent columnar memory format for flat and hierarchical data, organized for efficient analytic operations on modern hardware. It also provides computational libraries and zero-copy streaming messaging and interprocess communication. Languages currently supported include C, C++, C#, Go, Java, JavaScript, MATLAB, Python, R, Ruby, and Rust.
Apache Arrow is columnar in memory data structure, allows applications to avoid unnecessary IO and accelerate analytical processing performance on modern CPUs and GPUs.
Conventionally, this is how to share data across different systems:
With obvious disadvantages:
Each system has its own internal memory format 70–80% computation wasted on serialization and deserialization
Similar functionality implemented in multiple projects
With Apache Arrow as common data structures:
With obvious advantages when it comes to data transport amongst different systems that use Arrow memory data structure framework:
All systems utilize the same memory format No overhead for cross-system communication Projects can share functionality (eg, Parquet-to-Arrow reader)

Apache Arrow Flight

A Framework for Fast Data Transport Flight initially is focused on optimized transport of the Arrow columnar format (i.e. "Arrow record batches") over gRPC, Google’s popular HTTP/2-based general-purpose RPC library and framework. While we have focused on integration with gRPC, as a development framework Flight is not intended to be exclusive to gRPC.
One of the biggest features that sets apart Flight from other data transport frameworks is parallel transfers, allowing data to be streamed to or from a cluster of servers simultaneously. This enables developers to more easily create scalable data services that can serve a growing client base.
A simple Flight setup might consist of a single server to which clients connect and make DoGet requests.
Flight Basics
The Arrow Flight libraries provide a development framework for implementing a service that can send and receive data streams. A Flight server supports several basic kinds of requests:
1
Handshake: a simple request to determine whether the client is authorized and, in some cases, to establish an implementation-defined session token to use for future requests
2
ListFlights: return a list of available data streams
3
GetSchema: return the schema for a data stream
4
GetFlightInfo: return an "access plan" for a dataset of interest, possibly requiring consuming multiple data streams. This request can accept custom serialized commands containing, for example, your specific application parameters.
5
DoGet: send a data stream to a client
6
DoPut: receive a data stream from a client
7
DoAction: perform an implementation-specific action and return any results, i.e. a generalized function call
8
ListActions: return a list of available action types Also take advantage of gRPC’s elegant "bidirectional" streaming support (built on top of HTTP/2 streaming) to allow clients and servers to send data and metadata to each other simultaneously while requests are being served.
Copied!

Dremio

Dremio is the data lake engine
www.dremio.com
Dremio’s query engine is built on Apache Arrow, which is an in memory columnar data structure. Its SQL engine allows you to use SQL to query structured data such as relational database tables or non-structure such as key value pairs entities such as JSON, it is a distributed/clustered and in memory columnar query engine, that can run on one node or many nodes.

Dremio Flight Connector

Dremio Flight Connector is an implementation of Apache Arrow Flight Framework that allows a client, such as a Java program or Python script to request data from Dremio server using Apache Arrow Flight protocol, that inherits the data transport Apache Arrow data structure.
This means you can get data from Dremio or any systems that use Apache Arrow as memory data cache without ODBC or JDBC and without serialization and deserialization overhead that come with both.
Below is the integration demonstration of Dremio Data Lake Engine Server, Dremio Flight Connector and Apache Machine Learning Spam Detection.

Setup:

In my last writing, I have built Dremio server open source version. I am going to leverage Dremio server already built for this writing.
First task, I need to build Dremio Flight Connector that allows getting data from Dremio via Apache Flight, which is Arrow Memory Data Transport Protocol.
(1). To start, git clone Dremio Flight Connector from GitHub:
GitHub - dremio-hub/dremio-flight-connector: Dremio Flight connector. Access Dremio using Arrow flight
GitHub
(2). Then follow build instruction in the readme, make sure your mvn version is up to date and simply run: mvn clean install or if you do not have current or recent version of mvn: mvnw clean install
(3). Once build complete, get the jar file in the target folder inside the folder created by git, such as:
dremio-flight-connector-0.11.0-SNAPSHOT-shaded.jar
and copy it to <dremio home dir>/jars/
(4). Then modify /conf/dremio-env
1
DREMIO_JAVA_SERVER_EXTRA_OPTS=’-Ddremio.flight.enabled=true -Ddremio.flight.parallel.enabled=true -Ddremio.flight.use-ssl=true -Ddremio.flight.port=47470 -Ddremio.flight.host=0.0.0.0'
Copied!
(5). Restart dremio, assume you are in <dremio home dir>,
1
bin/dremio β€” config ./conf start
Copied!
(6). To make sure Dremio flight connector is up and running:
(i) Check java process to make sure β€œ-Ddremio.flight.enabled=true” is inside the dremio command line
1
ps -ef | grep dremio | grep java | grep -v grep
2
​
3
dremio 5588 1 2 11:46 pts/1 00:05:12 /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.191.b12–1.el7_6.x86_64/jre/bin/java -Djava.util.logging.config.class=org.slf4j.bridge.SLF4JBridgeHandler -Djava.library.path=/opt/dremio/dremio-oss/distribution/server/target/dremio-community-4.1.8–202003120636020140–9c2a6b13/dremio-community-4.1.8–202003120636020140–9c2a6b13/lib -XX:+PrintGCDetails -XX:+PrintGCDateStamps -Xloggc:/opt/dremio/dremio-oss/distribution/server/target/dremio-community-4.1.8–202003120636020140–9c2a6b13/dremio-community-4.1.8–202003120636020140–9c2a6b13/log/server.gc -Ddremio.log.path=/opt/dremio/dremio-oss/distribution/server/target/dremio-community-4.1.8–202003120636020140–9c2a6b13/dremio-community-4.1.8–202003120636020140–9c2a6b13/log -Ddremio.plugins.path=/opt/dremio/dremio-oss/distribution/server/target/dremio-community-4.1.8–202003120636020140–9c2a6b13/dremio-community-4.1.8–202003120636020140–9c2a6b13/plugins -Xmx4096m -XX:MaxDirectMemorySize=8192m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/opt/dremio/dremio-oss/distribution/server/target/dremio-community-4.1.8–202003120636020140–9c2a6b13/dremio-community-4.1.8–202003120636020140–9c2a6b13/log -Dio.netty.maxDirectMemory=0 -DMAPR_IMPALA_RA_THROTTLE -DMAPR_MAX_RA_STREAMS=400 -Ddremio.flight.enabled=true -Ddremio.flight.parallel.enabled=true -Ddremio.flight.use-ssl=true -Ddremio.flight.port=47470 -Ddremio.flight.host=0.0.0.0 … com.dremio.dac.daemon.DremioDaemon
Copied!
(ii) Check dremio server.log file, /log/server.log, look for below:
1
2020–03–21 17:16:39,835 [main] INFO com.dremio.flight.FlightInitializer β€” set up flight plugin on port 47470 and host 0.0.0.0
Copied!
(iii) Lastly, to make sure port 47470 (flight connector) is listened by Dremio, by telnet:
1
telnet 10.0.2.15 47470
2
​
3
Trying 10.0.2.15…
4
Connected to 10.0.2.15.
5
Escape character is '^]'.
Copied!

Apache Spark as client of Dremio server:

When all are done, you are in business to continue your data science work.
The dataset used in this writing is SMSSpamCollection, which text messaging data, which has labeled whether a SMS text is a Spam (a Spam) or Ham (not a Spam).
Following is from the provider of this dataset as description:
The SMS Spam Collection v.1 (hereafter the corpus) is a set of SMS tagged messages that have been collected for SMS Spam research. It contains one set of SMS messages in English of 5,574 messages, tagged according being ham (legitimate) or spam.
First, I placed the data file SMSSpamCollection into HDFS folder /dremio
1
hdfs dfs -ls /dremio/
2
​
3
Found 3 items
4
-rw-r β€” r β€” 3 hadoop supergroup 477907 2020–03–21 11:07 /dremio/SMSSpamCollection
Copied!
The file is tab (\t) delimited.
In dremio UI, load up and catalog this file as data source so this file can be queried by SQL, as if it were a table.
Once dataset file can be loaded by Dremio, which can now supply to client, such as Python script with Dremio Flight Connector, without ODBC, following is the Python code does the following:
Import pyarrow and flight library
Connect to Dremio via Arrow flight connector, including authentication
Fetching data from SMSSpamCollection file SQL and loading to a Pandas dataframe
Starts Apache Spark and convert the Pandas dataframe to SparkSQL dataframe
Conduct machine learning model traing and testing with Apache Spark ML
LogisticRegression and NeiveBayes
Following is the Python code that connects to Dremio with Dremio Flight Connector and does ML with Apache Spark:
1
#Import needed Apache Arrow Python library for flight
2
from pyarrow import flight
3
import pyarrow as pa
4
#Define Dremio Client Authentication Methods
5
class HttpDremioClientAuthHandler(flight.ClientAuthHandler):
6
def __init__(self, username, password):
7
super(flight.ClientAuthHandler, self).__init__()
8
self.basic_auth = flight.BasicAuth(username, password)
9
self.token = None
10
def authenticate(self, outgoing, incoming):
11
auth = self.basic_auth.serialize()
12
outgoing.write(auth)
13
self.token = incoming.read()
14
def get_token(self):
15
return self.token
16
username = β€˜george’
17
password = β€˜<redacted>’
18
sql = β€˜β€™β€™select * from β€œfraud.dremio”.SMSSpamCollection’’’
19
#Connect to Dremio with flight connector on port 47470 mentions earlier in the writing
20
client = flight.FlightClient(β€˜grpc+tcp://10.0.2.15:47470’)
21
client.authenticate(HttpDremioClientAuthHandler(username, password))
22
#passing in SQL query statement to Dremio, execute and returns the data in pandas
23
#dataframe pdf
24
info = client.get_flight_info(flight.FlightDescriptor.for_command(sql))
25
reader = client.do_get(info.endpoints[0].ticket)
26
batches = []
27
import pandas as pd
28
while True:
29
try:
30
batch, metadata = reader.read_chunk()
31
batches.append(batch)
32
except StopIteration:
33
break
34
data = pa.Table.from_batches(batches)
35
pdf = data.to_pandas()
36
pdf.head()
37
A B
38
0 ham Go until jurong point, crazy.. Available only …
39
1 ham Ok lar… Joking wif u oni…
40
2 spam Free entry in 2 a wkly comp to win FA Cup fina…
41
3 ham U dun say so early hor… U c already then say…
42
4 ham Nah I don’t think he goes to usf, he lives aro…
43
import sys,os,os.path
44
os.environ[β€˜SPARK_HOME’]=’/opt/spark/’
45
import matplotlib.pyplot as plt
46
%matplotlib inline
47
import findspark
48
findspark.init()
49
import pyspark
50
from pyspark.sql import SparkSession
51
from pyspark.sql import SQLContext
52
spark = SparkSession.builder.getOrCreate()
53
sc=spark.sparkContext
54
#Convert the pandas dataframe to SparkSQL dataframe
55
sqlCtx = SQLContext(sc)
56
df = sqlCtx.createDataFrame(pdf)
57
df.show(5)
58
​
59
/*
60
Output:
61
+ β€” β€” + β€” β€” β€” β€” β€” β€” β€” β€” β€” β€” +
62
| A | B |
63
+ β€” β€” + β€” β€” β€” β€” β€” β€” β€” β€” β€” β€” +
64
| ham|Go until jurong p… |
65
| ham|Ok lar… Joking … |
66
|spam|Free entry in 2 a… |
67
| ham|U dun say so earl… |
68
| ham|Nah I don’t think… |
69
+ β€” β€” + β€” β€” β€” β€” β€” β€” β€” β€” β€” β€” +
70
only showing top 5 rows
71
​
72
*/
73
​
74
#Note: Spam is Spam, Ham is OK. Rename Column name A as status, B as feature
75
df = df.withColumnRenamed('A', 'status').withColumnRenamed('B', 'message')
76
​
77
#Encode status column to numeric: ham to 1.0 and spam to 0.
78
#All our fields need to be numeric for machine to learn, also rename the column #status to label
79
​
80
df.createOrReplaceTempView('temp')
81
df = spark.sql('select case status when "ham" then 1.0 else 0 end as label, message from temp')
82
​
83
df.show()
84
​
85
/*
86
Output:
87
+-----+--------------------+
88
|label| message|
89
+-----+--------------------+
90
| 1.0|Go until jurong p...|
91
| 1.0|Ok lar... Joking ...|
92
| 0.0|Free entry in 2 a...|
93
| 1.0|U dun say so earl...|
94
| 1.0|Nah I don't think...|
95
| 0.0|FreeMsg Hey there...|
96
| 1.0|Even my brother i...|
97
| 1.0|As per your reque...|
98
| 0.0|WINNER!! As a val...|
99
| 0.0|Had your mobile 1...|
100
| 1.0|I'm gonna be home...|
101
| 0.0|SIX chances to wi...|
102
| 0.0|URGENT! You have ...|
103
| 1.0|I've been searchi...|
104
| 1.0|I HAVE A DATE ON ...|
105
| 0.0|XXXMobileMovieClu...|
106
| 1.0|Oh k...i'm watchi...|
107
| 1.0|Eh u remember how...|
108
| 1.0|Fine if thatΒ’s th...|
109
| 0.0|England v Macedon...|
110
+-----+--------------------+
111
only showing top 20 rows
112
​
113
*/
114
​
115
#1 is OK, 0 is Spam
116
#Tokenize the messages Tokenization is the process of taking text (such as a sentence)
117
# and breaking it into individual terms (usually words). Let’s tokenize the messages
118
#and create a list of words of each message.
119
from pyspark.ml.feature import Tokenizer
120
tokenizer = Tokenizer(inputCol="message", outputCol="words")
121
wordsData = tokenizer.transform(df)
122
wordsData.show(3)
123
​
124
/*
125
Output:
126
​
127
+-----+--------------------+--------------------+
128
|label| message| words|
129
+-----+--------------------+--------------------+
130
| 1.0|Go until jurong p...|[go, until, juron...|
131
| 1.0|Ok lar... Joking ...|[ok, lar..., joki...|
132
| 0.0|Free entry in 2 a...|[free, entry, in,...|
133
+-----+--------------------+--------------------+
134
only showing top 3 rows
135
​
136
*/
137
​
138
#CountVectorizer converts a collection of text documents to vectors of token counts.
139
from pyspark.ml.feature import CountVectorizer
140
count = CountVectorizer (inputCol="words", outputCol="rawFeatures")
141
model = count.fit(wordsData)
142
featurizedData = model.transform(wordsData)
143
featurizedData.show(3)
144
​
145
/*
146
Output:
147
​
148
+-----+--------------------+--------------------+--------------------+
149
|label| message| words| rawFeatures|
150
+-----+--------------------+--------------------+--------------------+
151
| 1.0|Go until jurong p...|[go, until, juron...|(13587,[8,43,53,6...|
152
| 1.0|Ok lar... Joking ...|[ok, lar..., joki...|(13587,[5,76,409,...|
153
| 0.0|Free entry in 2 a...|[free, entry, in,...|(13587,[0,3,8,22,...|
154
+-----+--------------------+--------------------+--------------------+
155
only showing top 3 rows
156
#IDF reduces the features that often appear in the corpus. When using text as a feature,
157
#this usually improves performance because the most common,
158
#and therefore less important, words are weighted down
159
​
160
*/
161
​
162
from pyspark.ml.feature import IDF
163
idf = IDF(inputCol="rawFeatures", outputCol="features")
164
idfModel = idf.fit(featurizedData)
165
rescaledData = idfModel.transform(featurizedData)
166
rescaledData.select("label", "features").show(3) #Only needed to train
167
​
168
/*
169
Output:
170
​
171
+-----+--------------------+
172
|label| features|
173
+-----+--------------------+
174
| 1.0|(13587,[8,43,53,6...|
175
| 1.0|(13587,[5,76,409,...|
176
| 0.0|(13587,[0,3,8,22,...|
177
+-----+--------------------+
178
​
179
only showing top 3 rows
180
*/
181
​
182
#Randomly Split DataFrame into 80% Training (trainDF) and 20 Testing (testDF)
183
seed = 0 # random seed 0
184
trainDF, testDF = rescaledData.randomSplit([0.8,0.2],seed)
185
​
186
#Logistic regression classifier
187
#Logistic regression is a common method of predicting classification responses.
188
#A special case of a generalized linear model is the probability of predicting a #result.
189
#In spark.ml, logistic regression can be used to predict binary results
190
#by binomial logistic regression, or it can be used to predict multiple types of #results by using multiple logistic regression.
191
#Use the family parameter to choose between these two algorithms, or leave it unset #and Spark
192
#will infer the correct variable.
193
​
194
from pyspark.ml.classification import LogisticRegression
195
from pyspark.ml.evaluation import BinaryClassificationEvaluator
196
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
197
import numpy as np
198
​
199
lr = LogisticRegression(maxIter = 100)
200
model_lr = lr.fit(trainDF)
201
prediction_lr = model_lr.transform(testDF)
202
​
203
from pyspark.ml.evaluation import BinaryClassificationEvaluator
204
my_eval_lr = BinaryClassificationEvaluator(rawPredictionCol='prediction', labelCol='label', metricName='areaUnderROC')
205
my_eval_lr.evaluate(prediction_lr)
206
/*
207
0.8734030197444833
208
*/
209
​
210
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
211
my_mc_lr = MulticlassClassificationEvaluator(predictionCol='prediction', labelCol='label', metricName='f1')
212
my_mc_lr.evaluate(prediction_lr)
213
​
214
/*
215
0.9654997463216642
216
*/
217
​
218
my_mc_lr = MulticlassClassificationEvaluator(predictionCol='prediction', labelCol='label', metricName='accuracy')
219
my_mc_lr.evaluate(prediction_lr)
220
​
221
/*
222
0.967479674796748
223
*/
224
​
225
train_fit_lr = prediction_lr.select('label','prediction')
226
train_fit_lr.groupBy('label','prediction').count().show()
227
​
228
/*
229
Output:
230
​
231
+-----+----------+-----+
232
|label|prediction|count|
233
+-----+----------+-----+
234
| 0.0| 1.0| 31|
235
| 1.0| 1.0| 860|
236
| 0.0| 0.0| 92|
237
| 1.0| 0.0| 1|
238
+-----+----------+-----+
239
*/
240
​
241
#Naive Bayes Naive Bayesian classifiers are a class of simple probability classifiers
242
#that apply strong (naive)
243
#independent assumptions between features based on Bayes' theorem.
244
#The spark.ml implementation
245
#currently supports polynomial naive Bayes and Bernoulli NaΓ―ve Bayes.
246
​
247
from pyspark.ml.classification import NaiveBayes
248
nb = NaiveBayes()
249
Model_nb = nb.fit(trainDF)
250
predictions_nb = Model_nb.transform(testDF)
251
​
252
predictions_nb.select(β€˜label’, β€˜prediction’).show(5)
253
​
254
/*
255
Output:
256
​
257
+-----+----------+
258
|label|prediction|
259
+-----+----------+
260
| 0.0| 0.0|
261
| 0.0| 0.0|
262
| 0.0| 0.0|
263
| 0.0| 0.0|
264
| 0.0| 0.0|
265
+-----+----------+
266
only showing top 5 rows
267
*/
268
​
269
from pyspark.ml.evaluation import BinaryClassificationEvaluator
270
my_eval_nb = BinaryClassificationEvaluator(rawPredictionCol='prediction', labelCol='label', metricName='areaUnderROC')
271
my_eval_nb.evaluate(predictions_nb)
272
​
273
/*
274
0.937862950058072
275
*/
276
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
277
my_mc_nb = MulticlassClassificationEvaluator(predictionCol='prediction', labelCol='label', metricName='f1')
278
my_mc_nb.evaluate(predictions_nb)
279
/*
280
0.933544453535483
281
*/
282
my_mc_nb = MulticlassClassificationEvaluator(predictionCol='prediction', labelCol='label', metricName='accuracy')
283
my_mc_nb.evaluate(predictions_nb)
284
/*
285
0.9278455284552846
286
*/
287
​
Copied!

Summary:

Following is the model metrics:
In this exercise, LogisticRegression appears similar to NaiveBayes. The objective of this writing is to demonstrate how to use Dremio Flight Connector to get data from Dremio server with Apache Arrow Flight protocol without ODBC or JDBC to run application with Apache Spark such as Machine Learning with Spark.
As always, code used in this writing is in my GitHub repo.
GitHub - geyungjen/jentekllc: Apache Spark Application Development -- George Jen, Jen Tek LLC
GitHub
​
Last modified 1yr ago