标签: apache-spark-sql

df[100, c("column")]在scala spark数据框中是否有任何替代方法.我想从一列火花数据框中选择特定的行.例如100th,在R等效代码中的行

val columnName=Seq("col1","col2",....."coln");

有没有办法执行dataframe.select操作以获取仅包含指定列名的数据帧.我知道我可以做,dataframe.select("col1","col2"...) 但是columnName在运行时生成.我可以dataframe.select()在循环中为每个列名重复执行.它会有任何性能开销吗？有没有其他更简单的方法来实现这一目标？

我有一个具有以下结构的数据帧:

 |-- data: struct (nullable = true)
 |    |-- id: long (nullable = true)
 |    |-- keyNote: struct (nullable = true)
 |    |    |-- key: string (nullable = true)
 |    |    |-- note: string (nullable = true)
 |    |-- details: map (nullable = true)
 |    |    |-- key: string
 |    |    |-- value: string (valueContainsNull = true)

如何展平结构并创建新的数据框:

     |-- id: long (nullable = true)
     |-- keyNote: struct (nullable = true)
     |    |-- key: string (nullable = true)
     |    |-- note: …

我正在使用包装函数将一些数据加载到Spark中:

def load_data( filename ):
    df = sqlContext.read.format("com.databricks.spark.csv")\
        .option("delimiter", "\t")\
        .option("header", "false")\
        .option("mode", "DROPMALFORMED")\
        .load(filename)
    # add the filename base as hostname
    ( hostname, _ ) = os.path.splitext( os.path.basename(filename) )
    ( hostname, _ ) = os.path.splitext( hostname )
    df = df.withColumn('hostname', lit(hostname))
    return df

具体来说,我使用glob来一次加载一堆文件:

df = load_data( '/scratch/*.txt.gz' )

文件是:

/scratch/host1.txt.gz
/scratch/host2.txt.gz
...

我想列"主机名"实际上包含文件的真实名称被加载,而不是水珠(即host1,host2等等,而不是*).

我怎样才能做到这一点？

我有一个largeDataFrame(多列和数十亿行)和一个smallDataFrame(单列和10,000行).

我想所有的行从过滤largeDataFrame每当some_identifier列在largeDataFrame比赛中的行之一smallDataFrame.

这是一个例子:

largeDataFrame

some_idenfitier,first_name
111,bob
123,phil
222,mary
456,sue

smallDataFrame

some_identifier
123
456

desiredOutput

111,bob
222,mary

这是我丑陋的解决方案.

val smallDataFrame2 = smallDataFrame.withColumn("is_bad", lit("bad_row"))
val desiredOutput = largeDataFrame.join(broadcast(smallDataFrame2), Seq("some_identifier"), "left").filter($"is_bad".isNull).drop("is_bad")

有更清洁的解决方案吗？

我有一个DataFrame与Timestamp列,我需要为转换Date格式.

是否有可用的Spark SQL函数？

虽然通过在星火JDBC连接获取来自SQL Server的数据,我发现我可以设置一些并行的参数,如partitionColumn,lowerBound,upperBound,和numPartitions.我已经通过spark文档,但无法理解它.

谁能解释一下这些参数的含义？

我想在DataSet中为Row类型编写一个编码器,用于我正在进行的地图操作.基本上,我不明白如何编写编码器.

以下是地图操作的示例:

In the example below, instead of returning Dataset<String>, I would like to return Dataset<Row>

Dataset<String> output = dataset1.flatMap(new FlatMapFunction<Row, String>() {
            @Override
            public Iterator<String> call(Row row) throws Exception {

                ArrayList<String> obj = //some map operation
                return obj.iterator();
            }
        },Encoders.STRING());

我明白,编码器需要编写如下代码:

    Encoder<Row> encoder = new Encoder<Row>() {
        @Override
        public StructType schema() {
            return join.schema();
            //return null;
        }

        @Override
        public ClassTag<Row> clsTag() {
            return null;
        }
    };

但是,我不理解编码器中的clsTag(),我试图找到一个可以演示相似内容的运行示例(即行类型的编码器)

编辑 - 这不是所提问题的副本:尝试将数据帧行映射到更新行时编码器错误,因为答案谈到在Spark 2.x中使用Spark 1.x(我不是这样做),我也在寻找用于Row类的编码器而不是解决错误.最后,我一直在寻找Java解决方案,而不是Scala.

调用collect()RDD会将整个数据集返回给驱动程序,这会导致内存不足,我们应该避免这种情况.

collect()如果在数据帧上调用,它的行为方式会相同吗？方法怎么
样select()？
它是否也像collect()在数据帧上调用一样工作？

我们正在从MongoDB读取数据Collection.Collection列有两个不同的值(例如:) (bson.Int64,int) (int,float).

我试图使用pyspark获取数据类型.

我的问题是有些列有不同的数据类型.

假设quantity并且weight是列

quantity           weight
---------          --------
12300              656
123566000000       789.6767
1238               56.22
345                23
345566677777789    21

实际上我们没有为mongo集合的任何列定义数据类型.

当我从中查询计数时 pyspark dataframe

dataframe.count()

我这样的例外

"Cannot cast STRING into a DoubleType (value: BsonString{value='200.0'})"