1

更新 事实证明，这与 Databricks Spark CSV 阅读器创建 DataFrame 的方式有关。在下面这个不起作用的示例中，我使用 Databricks CSV 阅读器读取人员和地址 CSV，然后将生成的 DataFrame 以 Parquet 格式写入 HDFS。

我更改了代码以创建 DataFrame：（与 people.csv 类似） 

JavaRDD<Address> address = context.textFile("/Users/sfelsheim/data/address.csv").map(
            new Function<String, Address>() {
                public Address call(String line) throws Exception {
                    String[] parts = line.split(",");

                    Address addr = new Address();
                    addr.setAddrId(parts[0]);
                    addr.setCity(parts[1]);
                    addr.setState(parts[2]);
                    addr.setZip(parts[3]);

                    return addr;
                }
            });

然后将生成的 DataFrame 以 Parquet 格式写入 HDFS，连接按预期工作

在这两种情况下，我都在阅读完全相同的 CSV。

尝试对从 HDFS 上的两个不同 parquet 文件创建的两个 DataFrame 执行简单连接时遇到问题。

[main] INFO org.apache.spark.SparkContext -运行 Spark 版本 1.4.1

从Hadoop 2.7.0使用 HDFS

这是一个示例来说明。

 public void testStrangeness(String[] args) {
    SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("joinIssue");
    JavaSparkContext context = new JavaSparkContext(conf);
    SQLContext sqlContext = new SQLContext(context);

    DataFrame people = sqlContext.parquetFile("hdfs://localhost:9000//datalake/sample/people.parquet");
    DataFrame address = sqlContext.parquetFile("hdfs://localhost:9000//datalake/sample/address.parquet");

    people.printSchema();
    address.printSchema();

    // yeah, works
    DataFrame cartJoin = address.join(people);
    cartJoin.printSchema();

    // boo, fails 
    DataFrame joined = address.join(people,
            address.col("addrid").equalTo(people.col("addressid")));

    joined.printSchema();
}

人的内容

first,last,addressid 
your,mom,1 
fred,flintstone,2

地址内容

addrid,city,state,zip
1,sometown,wi,4444
2,bedrock,il,1111

people.printSchema();

结果是...

root
 |-- first: string (nullable = true)
 |-- last: string (nullable = true)
 |-- addressid: integer (nullable = true)

address.printSchema();

结果是...

root
 |-- addrid: integer (nullable = true)
 |-- city: string (nullable = true)
 |-- state: string (nullable = true)
 |-- zip: integer (nullable = true)


DataFrame cartJoin = address.join(people);
cartJoin.printSchema();

笛卡尔连接工作正常， printSchema() 结果...

root
 |-- addrid: integer (nullable = true)
 |-- city: string (nullable = true)
 |-- state: string (nullable = true)
 |-- zip: integer (nullable = true)
 |-- first: string (nullable = true)
 |-- last: string (nullable = true)
 |-- addressid: integer (nullable = true)

这个加盟...

DataFrame joined = address.join(people,
address.col("addrid").equalTo(people.col("addressid")));

导致以下异常。

Exception in thread "main" org.apache.spark.sql.AnalysisException: **Cannot resolve column name "addrid" among (addrid, city, state, zip);**
    at org.apache.spark.sql.DataFrame$$anonfun$resolve$1.apply(DataFrame.scala:159)
    at org.apache.spark.sql.DataFrame$$anonfun$resolve$1.apply(DataFrame.scala:159)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.sql.DataFrame.resolve(DataFrame.scala:158)
    at org.apache.spark.sql.DataFrame.col(DataFrame.scala:558)
    at dw.dataflow.DataflowParser.testStrangeness(DataflowParser.java:36)
    at dw.dataflow.DataflowParser.main(DataflowParser.java:119)

我尝试更改它，以便人员和地址具有共同的关键属性（addressid）并使用..

address.join(people, "addressid");

但是得到了同样的结果。

有任何想法吗？？

谢谢

4

2 回答 2

2

原来问题是 CSV 文件是带有 BOM 的 UTF-8 格式。DataBricks CSV 实现不处理带有 BOM 的 UTF-8。在没有BOM的情况下将文件转换为 UTF-8 ，一切正常。

于 2015-09-10T15:50:53.117 回答
0

能够通过使用记事本++来解决这个问题。在“编码”菜单下，我将其从“以 UTF-8 BOM 编码”切换为“以 UTF-8 编码”。

于 2017-03-10T15:48:09.497 回答