关于sparksql中设置自定义自增列的相关要点（工作共踩过的坑-1）

?

这个方法有一个缺点：序列是从0开始的，monotonically_increasing_id函数无法接受参数，所以我们无法用其根据我们的业务进行指定序列。
所以，有一个想法于是去看了一下该方法的源码，发下如下特点：

　　首先看到函数的定义def monotonically_increasing_id(): Column = withExpr { MonotonicallyIncreasingID() }
　　深入查看MonotonicallyIncreasingID() ，具体源码如下：

private[sql] case class MonotonicallyIncreasingID() extends LeafExpression with Nondeterministic {

  /**
   * Record ID within each partition. By being transient,count‘s value is reset to 0 every time
   * we serialize and deserialize and initialize it.
   */
  @transient private[this] var count: Long = _

  @transient private[this] var partitionMask: Long = _

  override protected def initInternal(): Unit = {
    count = 0L
    partitionMask = TaskContext.getPartitionId().toLong << 33
  }

  override def nullable: Boolean = false

  override def dataType: DataType = LongType

  override protected def evalInternal(input: InternalRow): Long = {
    val currentCount = count
    count += 1
    partitionMask + currentCount
  }

  override def genCode(ctx: CodeGenContext,ev: GeneratedExpressionCode): String = {
    val countTerm = ctx.freshName("count")
    val partitionMaskTerm = ctx.freshName("partitionMask")
    ctx.addMutableState(ctx.JAVA_LONG,countTerm,s"$countTerm = 0L;")
    ctx.addMutableState(ctx.JAVA_LONG,partitionMaskTerm,
      s"$partitionMaskTerm = ((long) org.apache.spark.TaskContext.getPartitionId()) << 33;")

    ev.isNull = "false"
    s"""
      final ${ctx.javaType(dataType)} ${ev.value} = $partitionMaskTerm + $countTerm;
      $countTerm++;
    """
  }
}

我们可以发现这个类中重写了父类的initInternal()方法，指定了初始值count=0L，enmm这样子的话我们可不可以通过复写该类中的初始值来满足我们的业务需求

override protected def initInternal(): Unit = {
    count = 0L
    partitionMask = TaskContext.getPartitionId().toLong << 33
  }

（别想太多，一个业务涉及那么多序列，总不能用一次改一次吧，当然如果技术过硬，自己写一套方法以及类，用来接收参数1：序列起始值，参数2：序列终止值。当前技术不够且加班 导致这个想法凉凉）
方法二：rdd算子中的zipWithIndex（）方法
代码demo如下：

val dfEmp=sqlContext.read.options(conUtil.con("TEST_EMP"))
  .format("jdbc").load()
  .select("ENAME","EMPNO")
//对读取的dfEmp进行schema加列操作，增加一列且指定列数据类型
val schma=dfEmp.schema.add(StructField("TEST_NO",LongType))

val temp=dfEmp.rdd.zipWithIndex()
//可以在row中指定我们自己业务需求的序列初始值
val changed= temp.map(t => Row.merge(t._1,Row(t._2+340000000)))
val in=sqlContext.createDataFrame(changed,schma)
JdbcUtils.saveTable(in,properties)
结果如下所示：

?到此，入职的第一个坑填好了！貌似方法二还能够用zipWithUniqueId()方法进行实现，由于时间不够就没有一一的尝试了，如果各位小伙伴们有空可以尝试一下！　同时，如果小伙伴们有更加好的方法，求分享！求指导！感谢！！！！！　　欢迎留言！！！！

（编辑：厦门网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!

采用import org.apache.spark.sql.functions.中的

monotonically_increasing_id函数。
使用demo如下：
//从数据库中加载表TEST_EMP进入内存，并且取ENAME和EMPNO两列

val dfEmp=sqlContext.read.options(conUtil.con("TEST_EMP"))
      .format("jdbc").load()
      .select("ENAME","EMPNO")
val test =dfEmp
      .withColumn("TEST_NO",monotonically_increasing_id)
//向oracle中写数据，这个函数的使用前提是需要确定表"EMP_TMP"存在。且向这张表写入数据的时候最好字段进行对应，如果列多余数据库中的列数则会出现参数过多的错误。

JdbcUtils.saveTable(test,url,"EMP_TMP",properties)

//代码结果如下所示，在数据库中生成了一个从0开始自增的列