H2o-spark

1.满足下面条件：

• Linux/OS X/Windows
• Java 8+
• Python 2.7+ For Python version of Sparkling Water (PySparkling)
• Spark 2.3 and SPARK_HOME shell variable must point to your local Spark installation

 

2.安装对应版本的h2o-sparkling 

2.3.0为对应已安装的spark版本，保持版本一致，以防报错。

pip install h2o_pysparkling_2.3==2.3.0

3.完整代码：

from pysparkling import *
from h2o.grid.grid_search import H2OGridSearch
from h2o.estimators.gbm import H2OGradientBoostingEstimator

#外部模式：不依赖spark集群，不依赖spark集群资源计算，必须关闭spark.dynamicAllocation.enabled=false conf = H2OConf(ss).set_external_cluster_mode().set_user_name('jerry').use_auto_cluster_start().set_num_of_external_h2o_nodes(1).set('HADOOP_USER_NAME','jerry').set_h2o_cluster("10.111.23.70",54321).set_client_ip("10.111.23.70").set_h2o_driver_path("/home/dp/h2odriver/h2odriver-sw2.3.18-hdp2.6-extended.jar").set_mapper_xmx("2G").set_yarn_queue("default").set_cloud_name("h2o_gbm") 

#内部模式：依赖spark集群，h2o集群将作业分发在spark集群中，依赖spark集群资源计算

#二者任选其一

#保持默认设置即可

conf = H2OConf(ss)
conf.set_num_h2o_workers(2)

创建H2OContext
hc = H2OContext.getOrCreate(ss, conf)

if df_count \
        and df_count > 30000:
    percent = 30000 / df_count
    # df = df.randomSplit(weights=[1-percent,percent],seed=1234)[1]
    df = df.sample(fraction=percent, seed=1234)

h2o_df = hc.as_h2o_frame(df, framename="df_h2o")

y_lable_h2o_df = h2o_df[self.y_col]
rest_h2o_df = h2o_df.drop(self.y_col)

# 获取后字符列集合，将列值转换为枚举
h2o_rest_allcols_no_label = rest_h2o_df.columns

# --------------------------对字符列进行编码----------------------
# 获取字符列h2o DataFame
str_h2o_df = None
# 获取字符列名集合
str_list = list()
if len(h2o_df.types) > 0:
    # 获取字符列名集合
    str_list = [k for k, v in rest_h2o_df.types.items() if v == 'string']

    if len(str_list) > 0:
        # 如果字符列集合不为空，则将字符列h2o DataFame的每列值进行枚举化
        str_h2o_df = rest_h2o_df[str_list].asfactor()

# 将非空的字符列集合h2o DataFame 与剩余的数值列后h2o DataFrame,进行拼接
if str_h2o_df:
    # 找到剩余数值列集合
    rest_list = [i for i in h2o_rest_allcols_no_label if i not in str_list]

    # 如数值列集合存在，与（编码后的）字符列集合合并
    if len(rest_list) > 0:
        rest_h2o_df = rest_h2o_df[rest_list].concat(str_h2o_df)

    # 否则 将字符列集合赋值给剩余(不带GEO_Y) h2o_df:此时只有spark df只有字符列变量
    else:
        rest_h2o_df = str_h2o_df

# 否则 将只有数值列集合赋值给剩余(不带GEO_Y) h2o_df:此时只有spark df只有数值列
# 不用再赋值，因为删除了GEO_Y后，全部是无需编码的数值列特征

# --------------------------对字符列进行编码----------------------

# 需要将标签值列 转换成 枚举类型（离散特征）
y_lable_h2o_df = y_lable_h2o_df.asfactor()
h2o_df = y_lable_h2o_df.concat(rest_h2o_df)

predictors = h2o_df.names[:]

# ratios = [0.6, 0.2]
frs = h2o_df.split_frame(ratios=[.7], seed=12345)
train = frs[0]
train.frame_id = "Train"
valid = frs[1]
valid.frame_id = "Validation"

if self.y_col \
        and self.y_col in predictors:
    predictors.remove(self.y_col)

# ======================modified =========================
# 'learn_rate': 0.1, 'max_depth': 15, 'ntrees': 100, 'stopping_metric': 'AUTO'}
# gridsearch  交叉验证参数组合

gbm_param_grid = {
    'learn_rate': 0.06,  # default = 0.1,
    'max_depth': [5, 9],
    # 'max_depth': [7],
    'stopping_metric': 'logloss',
}

​
gbm_grid1 = H2OGridSearch(model=H2OGradientBoostingEstimator,
                          hyper_params=gbm_param_grid
                          )

gbm_grid1.train(x=predictors, y=self.y_col, training_frame=train, validation_frame=valid, ntrees=100, seed=1)

model_rf = gbm_grid1.get_grid(sort_by='auc', decreasing=True)

# print('tmpaaa')
best_rf1 = model_rf.models[0]
# print(best_rf1)
# print('aaa')
best_rf1_perg1 = best_rf1.model_performance(valid)
print('================= 校验集表现指标 =================', best_rf1_perg1)
print('================= 最优参数组合 ================= ', model_rf.get_hyperparams_dict(model_rf.model_ids[0]))
# print(best_rf1.varimp())

importance_cols = dict()
for i in best_rf1.varimp():
    # ======================modified =========================

    # importance_cols.setdefault(i[0],i[3])
    # value = i[3]
    # if 'e' in str(i[3]):
    #     arr = str(i[3]).split('e')
    #     value = float(arr[0]) * math.pow(math.e, int(arr[1]))
    importance_cols.setdefault(i[0], i[3])

if len(importance_cols) > 0:
    no_nan_len = 0
    for k, v in importance_cols.items():
        if v > 0:
            no_nan_len = no_nan_len + 1

    cumulative_importance_threshold = 0
    cumulative_importance_list = list()
    for k, v in sorted(importance_cols.items(), key=lambda x: x[1], reverse=True):
        if cumulative_importance_threshold <= no_nan_len * threshold:
            cumulative_importance_list.append(k)
            cumulative_importance_threshold = cumulative_importance_threshold + 1

    self.low_importance_list = cumulative_importance_list

    # print('============重要度========',no_nan_len,'===== ',len(cumulative_importance_list))

initalImportanceList = importance_cols

return initalImportanceList

​