Fregata：机器学习

• Fregata（http://talkingdata.com）是一个基于[Apache Spark](http://spark.apache.org/)的轻量级、超快速的大规模机器学习库，它提供了Scala语言的高级API。

• 更高精度：针对各类问题，Fregata相比MLLib能够达到更高的准确率。

• 更高速度：对于广义线性模型，Fregata通常只需一个数据遍历周期即可收敛。对于10亿×10亿的数据集，使用内存缓存时Fregata可在1分钟内训练出广义线性模型；不使用内存缓存时则需约10分钟。一般而言，Fregata的速度是MLLib的10至100倍。

• 无参数优化：Fregata采用GSA随机梯度下降优化算法，无需手动调整学习率，因为我们找到了在训练过程中自动计算合适学习率的方法。当处理超高维问题时，Fregata会动态计算剩余内存以决定输出结果的稀疏程度，从而自动平衡准确性和效率。这两项特性使得Fregata可以作为不同问题下数据处理的标准模块。

• 更轻量：Fregata仅使用Spark的标准API，这使其能够快速且无缝地集成到大多数基于Spark的业务数据处理流程中。

架构

本文档介绍的是Fregata 0.1版本。

• core：主要实现基于GSA的独立算法，包括分类回归和聚类。
  • 分类：支持二分类和多分类。
  • 回归：后续发布。
  • 聚类：后续发布。
• spark：主要通过封装core.jar来实现基于Spark的大规模机器学习算法，并提供相应的算法接口。

Fregata支持Spark 1.x和2.x，以及Scala 2.10和2.11。

算法

• 万亿LR
• 万亿SoftMax
• 逻辑回归
• 组合特征逻辑回归
• SoftMax
• RDT

安装

可以通过Maven或SBT两种方式获取Fregata：

• Maven的pom.xml

    <dependency>
       <groupId>com.talkingdata.fregata</groupId>
        <artifactId>core</artifactId>
        <version>0.0.3</version>
    </dependency>
    <dependency>
        <groupId>com.talkingdata.fregata</groupId>
        <artifactId>spark</artifactId>
        <version>0.0.3</version>
    </dependency>

• SBT的build.sbt

    // 如果部署到本地Maven仓库，请添加
    // resolvers += Resolver.mavenLocal
    libraryDependencies += "com.talkingdata.fregata" % "core" % "0.0.3"
    libraryDependencies += "com.talkingdata.fregata" % "spark" % "0.0.3"

如果希望手动部署到本地Maven仓库，可按以下步骤操作：

git clone https://github.com/TalkingData/Fregata.git
cd Fregata
mvn clean package install

快速入门

假设您已经熟悉Spark，下面的示例展示了如何使用Fregata的逻辑回归，实验数据可以从LIBSVM数据集获取。

• 按照“下载”部分的内容，通过Maven或SBT将Fregata添加到项目中。
• 导入相关包：

	import fregata.spark.data.LibSvmReader
	import fregata.spark.metrics.classification.{AreaUnderRoc, Accuracy}
	import fregata.spark.model.classification.LogisticRegression
	import org.apache.spark.{SparkConf, SparkContext}

• 使用Fregata的LibSvmReader API加载训练数据：

    val (_, trainData)  = LibSvmReader.read(sc, trainPath, numFeatures.toInt)
    val (_, testData)  = LibSvmReader.read(sc, testPath, numFeatures.toInt)

• 利用训练数据构建逻辑回归模型：

    val model = LogisticRegression.run(trainData)

• 预测样本的得分：

    val pd = model.classPredict(testData)

• 使用AUC或其他指标评估模型预测质量：

    val auc = AreaUnderRoc.of( pd.map{
      case ((x,l),(p,c)) =>
        p -> l
    })

输入数据格式

Fregata的训练API需要RDD[(fregata.Vector, fregata.Num)]，而预测API则需要相同格式的数据，或者不含标签的RDD[fregata.Vector]。

	import breeze.linalg.{Vector => BVector , SparseVector => BSparseVector , DenseVector => BDenseVector}
	import fregata.vector.{SparseVector => VSparseVector }

	package object fregata {
	  type Num = Double
	  type Vector = BVector[Num]
	  type SparseVector = BSparseVector[Num]
	  type SparseVector2 = VSparseVector[Num]
	  type DenseVector = BDenseVector[Num]
	  def zeros(n:Int) = BDenseVector.zeros[Num](n)
	  def norm(x:Vector) = breeze.linalg.norm(x,2.0)
	  def asNum(v:Double) : Num = v
	}

• 如果数据格式为LibSvm，则可以直接使用Fregata的LibSvmReader.read() API：

	// sc是Spark Context
	// path是输入数据在HDFS上的位置
	// numFeatures是单个实例的特征数量
	// minPartitions是返回的指向输入数据的RDD的最小分区数
	read(sc:SparkContext, path:String, numFeatures:Int=-1, minPartition:Int=-1):(Int, RDD[(fregata.Vector, fregata.Num)])

• 否则需要进行一些数据构造：

  • 使用SparseVector：

      // indices是一个从0开始的数组，表示非零特征的索引
      // values是一个存储对应值的数组
      // length是每个实例的总特征数
      // label是该实例的标签
  
      // 带标签的输入数据
      sc.textFile(input).map{
          val indicies = ...
          val values   = ...
          val label    = ...
          ...
          (new SparseVector(indices, values, length).asInstanceOf[Vector], asNum(label))
      }
  
      // 不带标签的输入数据（仅用于预测API）
      sc.textFile(input).map{
          val indicies = ...
          val values   = ...
          ...
          new SparseVector(indices, values, length).asInstanceOf[Vector]
      }
  

  • 使用DenseVector：

      // datas是每个特征的数值
      // label是该实例的标签
  
      // 带标签的输入数据
      sc.textFile(input).map{
          val datas = ...
          val label = ...
          ...
          (new DenseVector(datas).asInstanceOf[Vector], asNum(label))
      }
  
      // 不带标签的输入数据（仅用于预测API）
      sc.textFile(input).map{
          val datas = ...
          ...
          new DenseVector(indices, values，length).asInstanceOf[Vector]
      }
  

邮件列表：

• yongjun.tian@tendcloud.com
• haijun.liu@tendcloud.com
• xiatian.zhang@tendcloud.com
• fan.yao@tendcloud.com

贡献者：

由TalkingData贡献。

Fregata 快速上手指南

Fregata 是一个基于 Apache Spark 的轻量级、超大规模机器学习库，提供 Scala 高级 API。相比 Spark MLLib，它在广义线性模型等场景下具有更高的精度和速度（通常快 10-100 倍），且无需手动调节学习率。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统: Linux / macOS / Windows (需配置好 Java 环境)
• JDK: Java 8 或更高版本
• Spark: 支持 Spark 1.x 和 2.x
• Scala: 支持 Scala 2.10 和 2.11
• 构建工具: Maven 或 SBT

安装步骤

您可以通过 Maven 或 SBT 将 Fregata 集成到项目中。目前主要版本为 0.0.3。

方式一：Maven 项目

在 pom.xml 文件中添加以下依赖：

<dependency>
   <groupId>com.talkingdata.fregata</groupId>
    <artifactId>core</artifactId>
    <version>0.0.3</version>
</dependency>
<dependency>
    <groupId>com.talkingdata.fregata</groupId>
    <artifactId>spark</artifactId>
    <version>0.0.3</version>
</dependency>

方式二：SBT 项目

在 build.sbt 文件中添加以下依赖：

// 如果部署到本地 maven 仓库，请取消下面注释
// resolvers += Resolver.mavenLocal
libraryDependencies += "com.talkingdata.fregata" % "core" % "0.0.3"
libraryDependencies += "com.talkingdata.fregata" % "spark" % "0.0.3"

方式三：本地源码编译安装

如果需要手动部署到本地 Maven 仓库，可执行以下命令：

git clone https://github.com/TalkingData/Fregata.git
cd Fregata
mvn clean package install

基本使用

以下示例展示如何使用 Fregata 进行 逻辑回归 (Logistic Regression) 训练与预测。假设您已熟悉 Spark 基础操作，且数据格式为 LIBSVM。

1. 导入必要的包

import fregata.spark.data.LibSvmReader
import fregata.spark.metrics.classification.{AreaUnderRoc, Accuracy}
import fregata.spark.model.classification.LogisticRegression
import org.apache.spark.{SparkConf, SparkContext}

2. 加载训练与测试数据

使用 Fregata 提供的 LibSvmReader 直接读取数据：

val (_, trainData)  = LibSvmReader.read(sc, trainPath, numFeatures.toInt)
val (_, testData)  = LibSvmReader.read(sc, testPath, numFeatures.toInt)

3. 训练模型

调用 LogisticRegression.run 即可自动完成训练（无需设置学习率）：

val model = LogisticRegression.run(trainData)

4. 进行预测

使用训练好的模型对测试数据进行评分预测：

val pd = model.classPredict(testData)

5. 评估模型效果

计算 AUC (Area Under Roc) 等指标来评估模型质量：

val auc = AreaUnderRoc.of( pd.map{
  case ((x,l),(p,c)) =>
    p -> l
})

提示：实验数据可从 LIBSVM Data 获取。若数据非 LIBSVM 格式，需参照官方文档构造 SparseVector 或 DenseVector 格式的 RDD。