filtre étincelle dataframe avec la ligne de champ qui est un tableau de chaînes de caractères

À l'aide de l'Étincelle de 1,5 et Scala 2.10.6

Je suis en train de filtrer un dataframe par l'intermédiaire d'un champ "tags" qui est un tableau de chaînes de caractères. Recherche de toutes les lignes qui ont le tag "privé".

val report = df.select("*")
  .where(df("tags").contains("private"))

:

Exception in thread "main" org.apache.spark.sql.AnalysisException:
ne peut pas résoudre "Contient(tags, privé)" en raison d'incompatibilité de type de données:
argument 1 exige de type chaîne de caractères, cependant, "tags" est de tableau
type.;

Est la méthode de filtrage mieux adapté?

Mise à JOUR:

les données proviennent de cassandra adaptateur, mais un minimum d'exemple qui montre ce que je suis en train de faire et aussi obtient l'erreur ci-dessus est:

  def testData (sc: SparkContext): DataFrame = {
    val stringRDD = sc.parallelize(Seq("""
      { "name": "ed",
        "tags": ["red", "private"]
      }""",
      """{ "name": "fred",
        "tags": ["public", "blue"]
      }""")
    )
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    import sqlContext.implicits._
    sqlContext.read.json(stringRDD)
  }
  def run(sc: SparkContext) {
    val df1 = testData(sc)
    df1.show()
    val report = df1.select("*")
      .where(df1("tags").contains("private"))
    report.show()
  }

Mise à JOUR: les balises de tableau peut être n'importe quelle longueur et le "privé" de la balise peut être dans n'importe quelle position

Mise à JOUR: une solution qui fonctionne: UDF

val filterPriv = udf {(tags: mutable.WrappedArray[String]) => tags.contains("private")}
val report = df1.filter(filterPriv(df1("tags")))
post échantillon de vos données et comment u r de la création de la df
Une option est de construire un UDF.
Eh bien, après avoir regardé le code source (depuis le scaladoc pour Column.contains dit seulement "Contient l'autre" qui n'est pas très instructif), je vois que Column.contains construit une instance de org.apache.spark.sql.catalyst.expressions.Contains qui dit "Une fonction qui renvoie true si la chaîne left contient la chaîne de caractères right". Il semble donc que df1("tags").contains ne peut pas faire ce que nous voulons faire dans ce cas. Je ne sais pas quelle autre solution à proposer. Il y a un ArrayContains également dans ...expressions mais Column ne semble pas en faire usage.
En effet, après avoir changé les données à des chaînes au lieu d'un tableau de chaînes de caractères, je trouve que la requête fonctionne.
J'ai eu un UDF de travail: val filterPriv = udf {(tags: mutable.WrappedArray[String]) => tags.contains("private")}; val report = df1.filter(filterPriv(df1("tags"))) toujours à la recherche de quelque chose de plus sympa, mais au moins je ne suis pas bloqué. thx!

OriginalL'auteur navicore | 2016-01-17

  1. 22

    Je pense que si vous utilisez where(array_contains(...)) il va travailler. Voici mon résultat:

    scala> import org.apache.spark.SparkContext
import org.apache.spark.SparkContext
scala> import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.DataFrame
scala> def testData (sc: SparkContext): DataFrame = {
|     val stringRDD = sc.parallelize(Seq
|      ("""{ "name": "ned", "tags": ["blue", "big", "private"] }""",
|       """{ "name": "albert", "tags": ["private", "lumpy"] }""",
|       """{ "name": "zed", "tags": ["big", "private", "square"] }""",
|       """{ "name": "jed", "tags": ["green", "small", "round"] }""",
|       """{ "name": "ed", "tags": ["red", "private"] }""",
|       """{ "name": "fred", "tags": ["public", "blue"] }"""))
|     val sqlContext = new org.apache.spark.sql.SQLContext(sc)
|     import sqlContext.implicits._
|     sqlContext.read.json(stringRDD)
|   }
testData: (sc: org.apache.spark.SparkContext)org.apache.spark.sql.DataFrame
scala>   
| val df = testData (sc)
df: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>]
scala> val report = df.select ("*").where (array_contains (df("tags"), "private"))
report: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>]
scala> report.show
+------+--------------------+
|  name|                tags|
+------+--------------------+
|   ned|[blue, big, private]|
|albert|    [private, lumpy]|
|   zed|[big, private, sq...|
|    ed|      [red, private]|
+------+--------------------+

    Remarque qu'il fonctionne si vous écrivez where(array_contains(df("tags"), "private")), mais si vous écrivez where(df("tags").array_contains("private")) (plus directement analogue à ce que vous avez écrit à l'origine), il échoue avec array_contains is not a member of org.apache.spark.sql.Column. En regardant le code source de Column, je vois des trucs pour gérer contains (construction d'un Contains exemple) mais pas array_contains. C'est peut-être un oubli.

    .select("*") n'est pas nécessaire => df.where(...) ...

    OriginalL'auteur Robert Dodier

  2. 1

    Vous pouvez utiliser ordinale à consulter le tableau json est pour, par exemple, dans votre cas df("tags")(0). Voici un exemple de travail

    scala> val stringRDD = sc.parallelize(Seq("""
|       { "name": "ed",
|         "tags": ["private"]
|       }""",
|       """{ "name": "fred",
|         "tags": ["public"]
|       }""")
|     )
stringRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[87] at parallelize at <console>:22
scala> import sqlContext.implicits._
import sqlContext.implicits._
scala> sqlContext.read.json(stringRDD)
res28: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>]
scala> val df=sqlContext.read.json(stringRDD)
df: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>]
scala> df.columns
res29: Array[String] = Array(name, tags)
scala> df.dtypes
res30: Array[(String, String)] = Array((name,StringType), (tags,ArrayType(StringType,true)))
scala> val report = df.select("*").where(df("tags")(0).contains("private"))
report: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>]
scala> report.show
+----+-------------+
|name|         tags|
+----+-------------+
|  ed|List(private)|
+----+-------------+
    merci. fonctionne si pos est fixe, mais il ne l'est pas. J'aurais fait le test des données un peu plus complexe, il peut être n'importe quel nombre de balises dans le tableau, la position est arbitraire.
    ensuite, ton code devrait fonctionner. vérifier ma mise à jour
    intéressant, je suis en manque de quelque chose, ressemble exactement à ce que je faisais, mais l'obtention de l'erreur. double vérification étincelle versions maintenant...
    c'est sur 1.5.4
    thx. Je dois être croisement de main quelque part. J'ai essayé 1.5.1 1.6 et val report = df.select("*").where(df("tags").contains("private")) me donne cette erreur dans l'orig post. creuser...

    OriginalL'auteur Aravind R. Yarram