apache-spark - saveastable - spark sql tutorial
Cómo implementar el incremento automático en spark SQL(PySpark) (1)
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
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")