nasıl tespit ediyorum

Question

Ben istenen sütun örneği json şeması için .select

örnek çağırmadan önce varsa söyleyebilir nasıl kıvılcım sql bir json dosyası, bir DataFrame oluşturduğunuzda

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

ama hasColumn için iyi bir işlev bulamıyor:

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

Bu yapmak istediğim şey bu.

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

Answer 1

Aslında sen bile sadece dataframe üzerinde diyebilirsin, sütunları kullanmak için seçin aramak gerekmez: Sütun bu biraz garip dizide olup olmadığını ele verdik en yakın test etmektir hasColumn yöntemi doğrudan dataframes üzerinde Sonra

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

Eğer c kullanılabilir, böylece kendisi

// 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 

Alternatif Pimp my kütüphane desen kullanarak örtük bir sınıf tanımlayabilirsiniz Bir kullanım o kadar:

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

Answer 2

bunun için Diğer seçenek df.columns ve potential_columns üzerine (bu durumda bir intersect olarak) bazı dizi isleme yapmak olacaktır.

// 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 

Ne yazık ki, yukarıdaki iç nesne senaryosu için işe yaramaz. Bunun için şemaya bakmanız gerekecek.

Ben böylece daha fazla iş yapmak zorunda kalacak o jenerik hale getirmek için, sizin potential_columns için tam sütun adları

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 

Bu yalnızca bir düzey derin gider değiştirmek için gidiyorum.

Answer 3

Sadece varsayalım ve Try ile başarısız olmasına izin verin. Sade ve basit ve destekler keyfi bir yuvalama:

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 

Veya daha da basit:

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 eşdeğeri: o Try içindeki ifadeyi değerlendirecek olarak

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 optimal değildir karar almadan önce.

büyük veri kümeleri için

, Scala aşağıda kullanın:

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

Answer 5

Normalde bu döndüren

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

olduğunu kullanmak Başka bir seçenek bir boolean