sparklingwater的介绍与实践（commandline）

    sparkling-water是将spark和h2o集成与一体的工具&＃xff0c;主要思想是利用h2o进行数据挖掘&＃xff0c;而利用进行数据处理和一部分计算&＃xff0c;具体架构如下&＃xff1a;

    我们可以从图中看到&＃xff0c;spark对源数据做了处理&＃xff0c;然后交给h2o进行建模&＃xff0c;在预测阶段也作为了计算引擎&＃xff0c; sparkling-water的牛逼之处在于使用了和spark的一样的数据结构&＃xff0c;这样在数据处理的时候可以十分灵活。

   我们在加载数据的时候&＃xff0c;既可以使用spark&＃xff0c;也可以使用h2o&＃xff0c;spark和h2o直接可以共享同样的数据结构&＃xff08;RDD&＃xff09;&＃xff0c;但是我们在进行进行数据挖掘&＃xff08;h2o只能使用后缀为.hex的文件&＃xff09;&＃xff0c;因此需要转换才能够进行计算。

共享rdd数据结构有非常多好处&＃xff0c;比如就可以利用spark进行数据的清洗&＃xff0c;好了&＃xff0c;我们直接来看一下怎么使用。

一、下载与安装

&＃xff08;1&＃xff09;官方提供的下载地址

但是官方提供的地址下载十分慢&＃xff0c;当然有VPN的另谈了&＃xff0c;这里我提供了我的网盘地址

&＃xff08;2&＃xff09;下载后上传到linux上进行解压

unzip sparkling-water-3.26.2-2.4.zip

&＃xff08;3&＃xff09;启动sparkling-water

找到解压路径下的sparkling-water的bin下的shell&＃xff0c;进行启动即可

./sparkling-shell

启动结果如下&＃xff1a;

[root&＃64;node11 bin]# ./sparkling-shell Using Spark defined in the SPARK_HOME&＃61;/opt/software/spark-2.4.3-bin-hadoop2.7 environmental property-----Spark master (MASTER) : local[*]Spark home (SPARK_HOME) : /opt/software/spark-2.4.3-bin-hadoop2.7H2O build version : 3.26.0.2 (yau)Sparkling Water version : 3.26.2-2.4Spark build version : 2.4.3Scala version : 2.11
----19/08/20 15:59:15 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... us applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://node11:4040
Spark context available as &＃39;sc&＃39; (master &＃61; local[*], app id &＃61; local-1566316763461).
Spark session available as &＃39;spark&＃39;.
Welcome to____ __/ __/__ ___ _____/ /___\ \/ _ \/ _ &＃96;/ __/ &＃39;_//___/ .__/\_,_/_/ /_/\_\ version 2.4.3/_/Using Scala version 2.11.12 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_221)
Type in expressions to have them evaluated.
Type :help for more information.scala> import org.apache.spark.h2o._
import org.apache.spark.h2o._
scala>

至此单机版的sparkling-water就可以使用了 

二、实践案例

  这里的案例是我根据官方提供的资料&＃xff08;https://github.com/h2oai/sparkling-water/tree/master/examples&＃xff09;进行操练的。

&＃xff08;1&＃xff09;导入spark.h2o的包

import org.apache.spark.h2o._

&＃xff08;2&＃xff09;初始化&＃xff0c;其实就是启动h2o

val hc &＃61; H2OContext.getOrCreate(spark)

 命令运行后结果如下&＃xff1a;

19/08/20 16:04:44 WARN internal.InternalH2OBackend: To avoid non-deterministic behavior of Spark broadcast-based joins,
we recommend to set &＃96;spark.sql.autoBroadcastJoinThreshold&＃96; property of SparkSession to -1.
E.g. spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
We also recommend to avoid using broadcast hints in your Spark SQL code.
19/08/20 16:04:44 WARN internal.InternalH2OBackend: Increasing &＃39;spark.locality.wait&＃39; to value 0 (Infinitive) as we need to ensure we run on the nodes with H2O
hc: org.apache.spark.h2o.H2OContext &＃61;Sparkling Water Context:* Sparkling Water Version: 3.26.2-2.4* H2O name: sparkling-water-root_local-1566316763461* cluster size: 1* list of used nodes:(executorId, host, port)------------------------(driver,node11,54321)------------------------Open H2O Flow in browser: http://192.168.12.137:54321 (CMD &＃43; click in Mac OSX)


&＃xff08;3&＃xff09;导入hc的包

import hc.implicits._

&＃xff08;4&＃xff09;导入spark的包

import spark.implicits._

&＃xff08;5&＃xff09;定义天气数据的路径

val weatherDataFile &＃61; "/opt/software/sparkling-water-3.26.2-2.4/examples/smalldata/chicago/Chicago_Ohare_International_Airport.csv"

注意的是&＃xff0c;这里最好是输入绝对路径&＃xff0c;官方提供的是相对路径&＃xff0c;自己要进行处理&＃xff0c;我是直接使用的绝对路径&＃xff0c;不然下一步加载数据会报找不到路径

org.apache.spark.sql.AnalysisException: Path does not exist: file:/opt/software/sparkling-water-3.26.2-2.4/bin/examples/smalldata/chicago/Chicago_Ohare_International_Airport.csv;at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary$1.apply(DataSource.scala:558)at org.apache.spark.sql.execution.datasources.DataSource$$anonfun$org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary$1.apply(DataSource.scala:545)at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)at scala.collection.immutable.List.foreach(List.scala:392)at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)at scala.collection.immutable.List.flatMap(List.scala:355)at org.apache.spark.sql.execution.datasources.DataSource.org$apache$spark$sql$execution$datasources$DataSource$$checkAndGlobPathIfNecessary(DataSource.scala:545)at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:359)at org.apache.spark.sql.DataFrameReader.loadV1Source(DataFrameReader.scala:223)at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:211)at org.apache.spark.sql.DataFrameReader.csv(DataFrameReader.scala:615)at org.apache.spark.sql.DataFrameReader.csv(DataFrameReader.scala:467)... 55 elided

&＃xff08;6&＃xff09;利用上面定义的路径进行加载数据

val weatherTable &＃61; spark.read.option("header", "true").option("inferSchema", "true").csv(weatherDataFile).withColumn("Date", to_date(&＃39;Date, "MM/dd/yyyy")).withColumn("Year", year(&＃39;Date)).withColumn("Month", month(&＃39;Date)) .withColumn("DayofMonth", dayofmonth(&＃39;Date))

&＃xff08;7&＃xff09;导入java的包

import java.io.File

&＃xff08;8&＃xff09;定义航空的路径

val dataFile &＃61; "/opt/software/sparkling-water-3.26.2-2.4/examples/smalldata/airlines/allyears2k_headers.zip"


&＃xff08;9&＃xff09;加载航空数据

val airlinesH2OFrame &＃61; new H2OFrame(new File(dataFile))

会给我们这样一个返回&＃xff1a;

airlinesH2OFrame: water.fvec.H2OFrame &＃61;
Frame key: allyears2k_headers.hexcols: 31rows: 43978chunks: 1size: 2154864


&＃xff08;10&＃xff09;将.hex文件转换成rdd

val airlinesTable &＃61; hc.asDataFrame(airlinesH2OFrame)

&＃xff08;11&＃xff09;利用spark进行数据的过滤

val flightsToORD &＃61; airlinesTable.filter(&＃39;Dest &＃61;&＃61;&＃61; "ORD")

&＃xff08;12&＃xff09;计算一下看看过滤后还有多少数据

flightsToORD.count

结果&＃xff1a;

9/08/20 16:24:08 WARN util.Utils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting &＃39;spark.debug.maxToStringFields&＃39; in SparkEnv.conf.
res0: Long &＃61; 2103


&＃xff08;13&＃xff09;利用spark进行rdd的join操作&＃xff08;合并表&＃xff09;

val joinedDf &＃61; flightsToORD.join(weatherTable, Seq("Year", "Month", "DayofMonth"))

&＃xff08;14&＃xff09;导包

import water.support.H2OFrameSupport._

&＃xff08;15&＃xff09;转换成.hex

val joinedHf &＃61; columnsToCategorical(hc.asH2OFrame(joinedDf), Array("Year", "Month", "DayofMonth"))

&＃xff08;16&＃xff09;导入深度学习的包

import _root_.hex.deeplearning.DeepLearning
import _root_.hex.deeplearning.DeepLearningModel.DeepLearningParameters
import _root_.hex.deeplearning.DeepLearningModel.DeepLearningParameters.Activation

&＃xff08;17&＃xff09;设置深度学习的参数

val dlParams &＃61; new DeepLearningParameters()
dlParams._train &＃61; joinedHf
dlParams._response_column &＃61; "ArrDelay"
dlParams._epochs &＃61; 5
dlParams._activation &＃61; Activation.RectifierWithDropout
dlParams._hidden &＃61; Array[Int](100, 100)

&＃xff08;18&＃xff09;训练模型

val dl &＃61; new DeepLearning(dlParams)
val dlModel &＃61; dl.trainModel.get

运行结果&＃xff1a;

dlModel: hex.deeplearning.DeepLearningModel &＃61;
Model Metrics Type: RegressionDescription: Metrics reported on full training framemodel id: DeepLearning_model_1566317084383_1frame id: frame_rdd_21_b16087b00dcb5349ed00b2f0a1964249MSE: 246.94397RMSE: 15.714451mean residual deviance: 246.94397mean absolute error: 9.7153425root mean squared log error: NaN
Variable Importances:Variable Relative Importance Scaled Importance PercentageDepDelay 1.000000 1.000000 0.020609NASDelay 0.953474 0.953474 0.019650Diverted 0.952912 0.952912 0.019639Cancelled 0.940236 0.940236 0.019378DayofMonth.12 0.929144 ...


&＃xff08;19&＃xff09;模型预测

val predictionsHf &＃61; dlModel.score(joinedHf)
val predictionsDf &＃61; hc.asDataFrame(predictionsHf)

&＃xff08;20&＃xff09;查看预测结果

predictionsDf.show

&＃43;-------------------&＃43;
| predict|
&＃43;-------------------&＃43;
| -14.28115203904661|
|-17.384369532025993|
|-15.648360659746515|
|-21.735323004320165|
|-0.4630290696992674|
| -9.351177667940217|
| 112.65659409295617|
| 30.161421574369385|
| 15.403270012684139|
| 170.8349751399989|
| 12.498370529294341|
| 147.3795710418184|
|-6.1483336982319585|
| 44.329600499888926|
| 17.50615431570487|
| 102.51282569095915|
| 7.4154391246514955|
| 9.09458182717221|
|-12.357870505795454|
|-14.798434263256837|
&＃43;-------------------&＃43;
only showing top 20 rows

官方提供的是利用Rstudio进行查看的&＃xff0c;我这样其实不科学&＃xff0c;因为只能查看最多20条数据