Spark 过滤器永远不会应用于 Java 中的 DataFrame答案

【问题标题】：Spark filters are never applied to DataFrame in JavaSpark 过滤器永远不会应用于 Java 中的 DataFrame
【发布时间】：2015-09-26 14:44:04
【问题描述】：

我是 Spark 的新手，我有一个从两个 Oracle 表中获取数据的查询。此类表必须由字段连接，该字段与下面的代码配合良好。但是，我需要像在 Oracle 的“where”子句中那样应用过滤器。例如，带上年龄在 25 到 50 岁之间的员工。我还必须应用 GroupBy 过滤器并使用 OrderBy 对最终结果进行排序。问题是唯一正确执行的操作是从表中检索所有数据以及它们之间的连接。其余的过滤器根本没有应用，我不知道为什么。你能帮我解决这个问题吗？我确定我遗漏了一些东西，因为没有得到编译错误。数据加载正常，但“where”子句似乎对数据没有任何影响，尽管有年龄在 25 到 50 岁之间的员工。非常感谢！

public static JavaRDD<Row> getResultsFromQuery(String connectionUrl) {

    JavaSparkContext sparkContext = new JavaSparkContext(new SparkConf()
            .setAppName("SparkJdbcDs").setMaster("local"));
    SQLContext sqlContext = new SQLContext(sparkContext);

    Map<String, String> options = new HashMap<>();
    options.put("driver", "oracle.jdbc.OracleDriver");
    options.put("url", connectionUrl);
    options.put("dbtable", "EMPLOYEE");

    DataFrameReader dataFrameReader = sqlContext.read().format("jdbc")
            .options(options);

    DataFrame dataFrameFirstTable = dataFrameReader.load();

    options.put("dbtable", "DEPARTMENT");

    dataFrameReader = sqlContext.read().format("jdbc").options(options);

    DataFrame dataFrameSecondTable = dataFrameReader.load();

    //JOIN. IT WORKS JUST FINE!!!

    DataFrame resultingDataFrame = dataFrameFirstTable.join(dataFrameSecondTable, 
            "DEPARTMENTID");


    //FILTERS. THEY DO NOT THROW ERROR, BUT ARE NOT APPLIED. RESULTS ARE ALWAYS THE SAME, WITHOUT FILTERS
    resultingDataFrame.where(resultingDataFrame.col("AGE").geq(25));
    resultingDataFrame.where(resultingDataFrame.col("AGE").leq(50));

    JavaRDD<Row> resultFromQuery = resultingDataFrame.toJavaRDD();

    //HERE I CONFIRM THAT THE NUMBER OF ROWS GOTTEN IS ALWAYS THE SAME, SO THE FILTERS DO NOT WORK.
    System.out.println("Number of rows "+resultFromQuery.count());

    return resultFromQuery;

}

【问题讨论】：

标签： java oracle apache-spark dataframe

【解决方案1】：

where 返回一个新的数据帧并且不会改变现有的数据帧，因此您需要存储输出：

DataFrame greaterThan25 = resultingDataFrame.where(resultingDataFrame.col("AGE").geq(25));
DataFrame lessThanGreaterThan = greaterThan25.where(resultingDataFrame.col("AGE").leq(50));
JavaRDD<Row> resultFromQuery = lessThanGreaterThan.toJavaRDD();

或者你可以直接链接它：

DataFrame resultingDataFrame = dataFrameFirstTable.join(dataFrameSecondTable, "DEPARTMENTID")
  .where(resultingDataFrame.col("AGE").geq(25))
  .where(resultingDataFrame.col("AGE").leq(50));

【讨论】：

贾斯汀，非常感谢您的回答！它确实适用于 AGE 范围。但是，当我使用 .groupBy 并尝试将结果保存在 DataFrame 中时，它表示检索到的数据必须存储在 GroupedData 对象中，当我这样做时效果很好。问题是我不知道如何迭代此类对象或如何将其转换为 JavaRDD ，这是该方法最终必须返回的内容。您或其他任何人对如何实现这一目标有任何想法吗？非常感谢您的关注和帮助。
我附上我用来分组数据的代码：GroupedData resultingGroupData = lessThanGreaterThan.groupBy(resultingDataFrame.col("DEPARTMENT_NAME")); //我应用 groupBy 命令 System.out.println("Number of data gotten: "+resultingGroupData .count().count()); //这里我确认它有 5 个组，这是数据库中存在的部门
您可以使用的方法可以在文档中找到。 spark.apache.org/docs/1.4.0/api/scala/… 你可能想使用 agg。或者，你可以下拉到一个 RDD 并使用那个 groupBy，它返回一个 RDD[(K,Iterable)]
非常感谢，贾斯汀！你帮了我很多。

【解决方案2】：

people.select("person_id", "first_name").filter(people("person_id") == 2).show

它不起作用，您将收到以下错误：

错误：使用替代方法重载方法值过滤器： (conditionExpr: String)org.apache.spark.sql.DataFrame （条件：org.apache.spark.sql.Column） org.apache.spark.sql.DataFrame 不能应用于（布尔）

看来，要使用 Spark 数据帧中的 Select 子句和过滤器，我们不能传递布尔值。

这两个查询用于从带有两个不同子句 where 和 filter 的 Spark DataFrame 中选择单行。

people.select("person_id", "first_name").filter(people("person_id") === 2).show

people.select("person_id", "first_name").where(people("person_id") === 2).show

使用上述查询之一，从 Spark DataFrame 中选择单行。

【讨论】：