apache-spark - saveastable - spark sql tutorial

Cómo implementar el incremento automático en spark SQL(PySpark)

Question

Necesito implementar una columna de incremento automático en mi tabla spark sql, ¿cómo podría hacerlo? Amablemente guíame estoy usando pyspark 2.0

Gracias Kalyan

Answer 1

Escribo / reutilizo Hive udf con estado y me registro con pySpark ya que Spark SQL tiene buen soporte para Hive.

compruebe esta línea @UDFType(deterministic = false, stateful = true) en el código siguiente para asegurarse de que sea UDF con estado.

package org.apache.hadoop.hive.contrib.udf;

import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.hive.ql.udf.UDFType;
import org.apache.hadoop.io.LongWritable;

/**
 * UDFRowSequence.
 */
@Description(name = "row_sequence",
    value = "_FUNC_() - Returns a generated row sequence number starting from 1")
@UDFType(deterministic = false, stateful = true)
public class UDFRowSequence extends UDF
{
  private LongWritable result = new LongWritable();

  public UDFRowSequence() {
    result.set(0);
  }

  public LongWritable evaluate() {
    result.set(result.get() + 1);
    return result;
  }
}

// End UDFRowSequence.java

Ahora construya el jar y agregue la ubicación cuando se inicie pyspark get.

$ pyspark --jars your_jar_name.jar

Entonces regístrese con sqlContext .

sqlContext.sql("CREATE TEMPORARY FUNCTION row_seq AS ''org.apache.hadoop.hive.contrib.udf.UDFRowSequence''")

Ahora usa row_seq() en la consulta de selección

sqlContext.sql("SELECT row_seq(), col1, col2 FROM table_name")

Proyecto para usar UDF de Hive en pySpark