jaki sposób wykryć, czy dataframe iskra ma kolumnę

Question

Kiedy tworzę DataFrame z pliku json w zapłonowej SQL, w jaki sposób mogę sprawdzić, czy dana kolumna występuje przed wywołaniem .select

przykład na przykład schematu json

{ 
    "a": { 
    "b": 1, 
    "c": 2 
    } 
} 

to jest to, co chcę zrobić:

potential_columns = Seq("b", "c", "d") 
df = sqlContext.read.json(filename) 
potential_columns.map(column => if(df.hasColumn(column)) df.select(s"a.$column")) 

ale nie mogę znaleźć dobrego funkcję hasColumn. Najbliższy stałam jest sprawdzenie, czy kolumna jest w tym nieco nietypowej tablicy:

scala> df.select("a.*").columns 
res17: Array[String] = Array(b, c) 

Answer 1

Właściwie nawet nie trzeba zadzwonić wybrać, aby korzystać z kolumn, można po prostu nazwać to na dataframe sam

// define test data 
case class Test(a: Int, b: Int) 
val testList = List(Test(1,2), Test(3,4)) 
val testDF = sqlContext.createDataFrame(testList) 

// define the hasColumn function 
def hasColumn(df: org.apache.spark.sql.DataFrame, colName: String) = df.columns.contains(colName) 

// then you can just use it on the DF with a given column name 
hasColumn(testDF, "a") // <-- true 
hasColumn(testDF, "c") // <-- false 

Alternatywnie można zdefiniować niejawny klasę za pomocą Pimp my biblioteki wzór tak, że metoda hasColumn jest dostępny na swoim dataframes bezpośrednio

implicit class DataFrameImprovements(df: org.apache.spark.sql.DataFrame) { 
    def hasColumn(colName: String) = df.columns.contains(colName) 
} 

Wtedy c używać go jako:

testDF.hasColumn("a") // <-- true 
testDF.hasColumn("c") // <-- false 

Answer 2

Twój innych opcji dla tego byłoby zrobić jakąś manipulację tablicy (w tym przypadku intersect) na df.columns i swojej potential_columns.

// Loading some data (so you can just copy & paste right into spark-shell) 
case class Document(a: String, b: String, c: String) 
val df = sc.parallelize(Seq(Document("a", "b", "c")), 2).toDF 

// The columns we want to extract 
val potential_columns = Seq("b", "c", "d") 

// Get the intersect of the potential columns and the actual columns, 
// we turn the array of strings into column objects 
// Finally turn the result into a vararg (: _*) 
df.select(potential_columns.intersect(df.columns).map(df(_)): _*).show 

Niestety to nie zadziała dla twojego scenariusza wewnętrznego obiektu powyżej. Będziesz musiał przyjrzeć się temu schematowi.

Zamierzam zmienić potential_columns w pełni kwalifikowane nazwy kolumn

val potential_columns = Seq("a.b", "a.c", "a.d") 

// Our object model 
case class Document(a: String, b: String, c: String) 
case class Document2(a: Document, b: String, c: String) 

// And some data... 
val df = sc.parallelize(Seq(Document2(Document("a", "b", "c"), "c2")), 2).toDF 

// We go through each of the fields in the schema. 
// For StructTypes we return an array of parentName.fieldName 
// For everything else we return an array containing just the field name 
// We then flatten the complete list of field names 
// Then we intersect that with our potential_columns leaving us just a list of column we want 
// we turn the array of strings into column objects 
// Finally turn the result into a vararg (: _*) 
df.select(df.schema.map(a => a.dataType match { case s : org.apache.spark.sql.types.StructType => s.fieldNames.map(x => a.name + "." + x) case _ => Array(a.name) }).flatMap(x => x).intersect(potential_columns).map(df(_)) : _*).show 

To tylko jeden poziom idzie głęboko, tak aby uczynić go generic trzeba by zrobić więcej pracy.

Answer 3

Po prostu załóż, że istnieje i pozwól mu zawieść z Try. Zwykły, prosty i obsługuje dowolną gniazdowania:

import scala.util.Try 
import org.apache.spark.sql.DataFrame 

def hasColumn(df: DataFrame, path: String) = Try(df(path)).isSuccess 

val df = sqlContext.read.json(sc.parallelize(
    """{"foo": [{"bar": {"foobar": 3}}]}""" :: Nil)) 

hasColumn(df, "foobar") 
// Boolean = false 

hasColumn(df, "foo") 
// Boolean = true 

hasColumn(df, "foo.bar") 
// Boolean = true 

hasColumn(df, "foo.bar.foobar") 
// Boolean = true 

hasColumn(df, "foo.bar.foobaz") 
// Boolean = false 

Albo jeszcze prościej:

val columns = Seq(
    "foobar", "foo", "foo.bar", "foo.bar.foobar", "foo.bar.foobaz") 

columns.flatMap(c => Try(df(c)).toOption) 
// Seq[org.apache.spark.sql.Column] = List(
// foo, foo.bar AS bar#12, foo.bar.foobar AS foobar#13) 

Python równoważne:

from pyspark.sql.utils import AnalysisException 
from pyspark.sql import Row 


def has_column(df, col): 
    try: 
     df[col] 
     return True 
    except AnalysisException: 
     return False 

df = sc.parallelize([Row(foo=[Row(bar=Row(foobar=3))])]).toDF() 

has_column(df, "foobar") 
## False 

has_column(df, "foo") 
## True 

has_column(df, "foo.bar") 
## True 

has_column(df, "foo.bar.foobar") 
## True 

has_column(df, "foo.bar.foobaz") 
## False 

Answer 4

Try nie jest optymalny, ponieważ będzie to ocena ekspresji wewnątrz Try zanim podejmie decyzję.

przypadku dużych zbiorów danych, użyj poniższej Scala:

df.schema.fieldNames.contains("column_name") 

Answer 5

Innym rozwiązaniem, które normalnie używam jest

df.columns.contains("column-name-to-check") 

zwraca Ten logiczną