¿Cómo comprobar si el marco de datos de Spark está vacío?

En este momento, tengo que usar df.count > 0para verificar si DataFrameestá vacío o no. Pero es algo ineficiente. ¿Hay alguna forma mejor de hacerlo?

Gracias.

PD: quiero verificar si está vacío para que solo guarde el DataFramesi no está vacío

Para Spark 2.1.0, mi sugerencia sería usar head(n: Int)o take(n: Int)con isEmpty, el que tenga la intención más clara para ti.

df.head(1).isEmpty
df.take(1).isEmpty

con el equivalente de Python:

len(df.head(1)) == 0  # or bool(df.head(1))
len(df.take(1)) == 0  # or bool(df.take(1))

El uso de df.first()y df.head()devolverá el java.util.NoSuchElementExceptionsi el DataFrame está vacío. first()llama head()directamente, que llama head(1).head.

def first(): T = head()
def head(): T = head(1).head

head(1)devuelve un Array, por lo que asumir headese Array hace java.util.NoSuchElementExceptionque el DataFrame esté vacío.

def head(n: Int): Array[T] = withAction("head", limit(n).queryExecution)(collectFromPlan)

Entonces, en lugar de llamar head(), use head(1)directamente para obtener la matriz y luego puede usar isEmpty.

take(n)también es equivalente a head(n)...

def take(n: Int): Array[T] = head(n)

Y limit(1).collect()es equivalente a head(1)(aviso limit(n).queryExecutionen el head(n: Int)método), por lo que los siguientes son todos equivalentes, al menos por lo que puedo decir, y no tendrá que detectar una java.util.NoSuchElementExceptionexcepción cuando el DataFrame esté vacío.

df.head(1).isEmpty
df.take(1).isEmpty
df.limit(1).collect().isEmpty

Sé que esta es una pregunta anterior, así que espero que ayude a alguien que use una versión más nueva de Spark.

Yo diría que solo tome el subyacente RDD. En Scala:

df.rdd.isEmpty

en Python:

df.rdd.isEmpty()

Dicho esto, todo lo que esto hace es llamar take(1).length, por lo que hará lo mismo que respondió Rohan ... ¿quizás un poco más explícito?

Puede aprovechar las funciones head()(o first()) para ver si DataFrametiene una sola fila. Si es así, no está vacío.

Si lo hace df.count > 0. Toma los recuentos de todas las particiones en todos los ejecutores y los suma en Driver. Esto lleva un tiempo cuando se trata de millones de filas.

La mejor manera de hacer esto es realizar df.take(1)y verificar si es nulo. Esto volverá java.util.NoSuchElementExceptionasí que es mejor intentarlo df.take(1).

El marco de datos devuelve un error cuando take(1)se realiza en lugar de una fila vacía. He resaltado las líneas de código específicas donde arroja el error.

Desde Spark 2.4.0 existe Dataset.isEmpty.

Su implementación es:

def isEmpty: Boolean = 
  withAction("isEmpty", limit(1).groupBy().count().queryExecution) { plan =>
    plan.executeCollect().head.getLong(0) == 0
}

Tenga en cuenta que a DataFrameya no es una clase en Scala, es solo un alias de tipo (probablemente cambiado con Spark 2.0):

type DataFrame = Dataset[Row]

Para los usuarios de Java, puede usar esto en un conjunto de datos:

public boolean isDatasetEmpty(Dataset<Row> ds) {
        boolean isEmpty;
        try {
            isEmpty = ((Row[]) ds.head(1)).length == 0;
        } catch (Exception e) {
            return true;
        }
        return isEmpty;
}

Esto verifica todos los escenarios posibles (vacío, nulo).

En Scala, puede usar implícitos para agregar los métodos isEmpty()y nonEmpty()la API de DataFrame, lo que hará que el código sea un poco más agradable de leer.

object DataFrameExtensions {
  implicit def extendedDataFrame(dataFrame: DataFrame): ExtendedDataFrame = 
    new ExtendedDataFrame(dataFrame: DataFrame)

  class ExtendedDataFrame(dataFrame: DataFrame) {
    def isEmpty(): Boolean = dataFrame.head(1).isEmpty // Any implementation can be used
    def nonEmpty(): Boolean = !isEmpty
  }
}

Aquí, también se pueden agregar otros métodos. Para usar la conversión implícita, use import DataFrameExtensions._en el archivo que desea usar la funcionalidad extendida. Posteriormente, los métodos se pueden utilizar directamente de la siguiente manera:

val df: DataFrame = ...
if (df.isEmpty) {
  // Do something
}

Tenía la misma pregunta y probé 3 soluciones principales:

1. df! = null df.count> 0
2. df.head (1) .isEmpty () como sugiere @ hulin003
3. df.rdd.isEmpty como sugiere @Justin Pihony

y, por supuesto, los 3 funcionan, sin embargo, en términos de rendimiento, esto es lo que encontré, al ejecutar estos métodos en el mismo DF en mi máquina, en términos de tiempo de ejecución:

1. se necesitan ~ 9366ms
2. se necesitan ~ 5607ms
3. toma ~ 1921ms
   

por lo tanto, creo que la mejor solución es df.rdd.isEmpty como sugiere @Justin Pihony

Encontré que en algunos casos:

>>>print(type(df))
<class 'pyspark.sql.dataframe.DataFrame'>

>>>df.take(1).isEmpty
'list' object has no attribute 'isEmpty'

esto es lo mismo para "longitud" o reemplace take () por head ()

[Solución] para el problema que podemos utilizar.

>>>df.limit(2).count() > 1
False

Si está utilizando Pypsark, también puede hacer:

len(df.head(1)) > 0

En PySpark, también puede utilizar este bool(df.head(1))para obtener una Truede Falsevalor

Devuelve Falsesi el marco de datos no contiene filas

df1.take(1).length>0

El takemétodo devuelve la matriz de filas, por lo que si el tamaño de la matriz es igual a cero, no hay registros en df.

dataframe.limit(1).count > 0

Esto también desencadena un trabajo, pero como estamos seleccionando un solo registro, incluso en el caso de miles de millones de registros de escala, el consumo de tiempo podría ser mucho menor.

De: https://medium.com/checking-emptiness-in-distributed-objects/count-vs-isempty-surprised-to-see-the-impact-fa70c0246ee0

Puedes hacerlo como:

val df = sqlContext.emptyDataFrame
if( df.eq(sqlContext.emptyDataFrame) )
    println("empty df ")
else 
    println("normal df")