Spark MLlib 开发实践

步骤一：依赖配置

Lindorm计算引擎JAR作业依赖社区版Spark 3.2.1版本，并且需要将scope字段配置为provided。示例如下：

<!-- Example -->
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-core_2.12</artifactId>
  <version>3.2.1</version>
  <scope>provided</scope>
</dependency>

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-mllib_2.12</artifactId>
  <version>3.2.1</version>
  <scope>provided</scope>
</dependency>

步骤二：参数配置

Lindorm计算引擎提供的配置项和配置方法请参见作业配置说明。

步骤三：代码示例

package com.alibaba.lindorm.ldps

import org.apache.spark.sql.SparkSession
import org.apache.spark.ml.regression.LinearRegression
import org.apache.spark.ml.linalg.Vectors

object SimpleMLlibExample {
  def main(args: Array[String]): Unit = {

    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("SimpleMLlibExample")
      .getOrCreate()

    // 构造训练数据 DataFrame
    val trainingData = Seq(
      (1.0, Vectors.dense(1.0)),
      (2.0, Vectors.dense(2.0)),
      (3.0, Vectors.dense(3.0)),
      (4.0, Vectors.dense(4.0))
    )

    import spark.implicits._
    val trainingDF = trainingData.toDF("label", "features")

    // 创建线性回归模型
    val lr = new LinearRegression()
      .setMaxIter(10)

    // 训练模型
    val model = lr.fit(trainingDF)

    // 打印模型系数
    println("Coefficients: " + model.coefficients)
    println("Intercept: " + model.intercept)

    // 测试数据
    val testDF = Seq(
      Tuple1(Vectors.dense(5.0))
    ).toDF("features")

    // 预测
    val predictions = model.transform(testDF)
    predictions.show()

    spark.stop()
  }
}

步骤四：作业提交

Lindorm计算引擎支持以下两种方式提交并管理作业。

• 通过Lindorm控制台提交作业，操作方法请参见通过控制台管理作业。

• 通过DMS提交作业，操作方法请参见通过DMS管理作业。

以下示例展示通过控制台提交任务。

通过控制台提交任务：

查看任务状态与结果输出：

输出结果：