标签: apache-spark-sql

请注意，这个问题是不是一个重复这一个！我不使用 Spark SQL 进行分区！我正在保存单个镶木地板文件！

我也在使用不支持 Hive-flavored SQL 的Databricks。

我在 Hive 中有一个表（我使用的是 Databricks），它包含两个分区。它看起来像这样：

CREATE TABLE foo_test (`col0` STRING, `col1` STRING, `col2` STRING, `datestamp` STRING)
USING parquet
OPTIONS (
  `serialization.format` '1'
)
PARTITIONED BY (datestamp)

编辑： *这也是调用的结果show create table foo_test;

我手动添加了两个带有 Spark 的分区到这个表中：

df = spark.read.csv(file_path.format(datestamp), header=True, inferSchema=True)

partitionPath = "/mnt/foo_test/datestamp={}/".format(datestamp)

df.coalesce(1).write.parquet(partitionPath, mode="overwrite")


/mnt/foo_test/datestamp=20180101/
/mnt/foo_test/datestamp=20180102/

如果我用 Spark 加载数据，我可以看到它在那里：

spark.read.option("mergeSchema", True).parquet("/mnt/foo_test").show()

+----+----+----+----+---------+
|col0|col1|col2|col3|datestamp|
+----+----+----+----+---------+
| foo| bar| baz|   1| 20180102|
| xul| qux| wom| …

我正在使用 DataFrame API 在 Spark 中编写以下代码。

val cond = "col("firstValue") >= 0.5 & col("secondValue") >= 0.5 & col("thirdValue") >= 0.5"
val Output1 = InputDF.where(cond)

我将所有条件作为来自外部参数的字符串传递，但它会抛出一个解析错误，因为它cond应该是类型Column。

例如：

col("firstValue") >= 0.5 & col("secondValue") >= 0.5 & col("thirdValue") >= 0.5

由于我想动态传递多个条件，如何将 a 转换String为 a Column？

编辑

有什么东西可以让我从外部读取条件列表 as Column，因为我没有找到任何可以使用 Scala 代码将 a 转换String为 a 的东西Column。

我有一列包含20180501字符串格式的数据，我想将其转换为日期格式，尝试使用

to_date(cast(unix_timestamp('20180501', 'YYYYMMDD') as timestamp))'

但仍然没有奏效。我将 Spark SQL 与数据帧一起使用

我正在查看 Spark 源代码以了解dropDuplicates方法的工作原理。在方法定义中有一个方法Deduplicate调用。但我找不到它的定义或参考。如果有人能指出我正确的方向，那就太好了。链接在这里。

我正在通过以下代码读取 csv 文件：-

    from pyspark.sql import SparkSession
    spark = SparkSession.builder \
            .master("local[2]") \
            .getOrCreate()

现在有四种不同的阅读选项：

1. df = spark.read.load("/..../xyz.csv")
2. df = spark.read.csv("/..../xyz.csv")
3. df = spark.read.format('csv').load("/..../xyz.csv")
4. df = spark.read.option().csv("/..../xyz.csv")

我应该使用哪个选项？

编辑：-

此外，无论是inferSchema="true"和inferSchema=True正在工作。我们可以盲目使用任何一种吗？

我正在使用 Pandas 读取 csv 文件，它是一个两列数据帧，然后我试图转换为 spark 数据帧。代码如下：

from pyspark.sql import SQLContext
sqlCtx = SQLContext(sc)
sdf = sqlCtx.createDataFrame(df)

数据框：

print(df) 

给出了这个：

    Name    Category
0   EDSJOBLIST apply at www.edsjoblist.com  ['biotechnology', 'clinical', 'diagnostic', 'd...
1   Power Direct Marketing  ['advertising', 'analytics', 'brand positionin...
2   CHA Hollywood Medical Center, L.P.  ['general medical and surgical hospital', 'hea...
3   JING JING GOURMET   [nan]
4   TRUE LIFE KINGDOM MINISTRIES    ['religious organization']
5   fasterproms ['microsoft .net']
6   STEREO ZONE ['accessory', 'audio', 'car audio', 'chrome', ...
7   SAN FRANCISCO NEUROLOGICAL …

我正在尝试删除一个创建了 Spark-Sql 的表（内部）表，一些表是如何被删除的，但表的位置仍然存在。有人可以让我知道如何做到这一点吗？

我尝试了 Beeline 和 Spark-Sql

    create table something(hello string)
    PARTITIONED BY(date_d string)
    ROW FORMAT DELIMITED FIELDS TERMINATED BY "^"
    LOCATION "hdfs://path"
    )
    Drop table something;
No rows affected (0.945 seconds)

谢谢

我使用以下结构构建了一个 spark DataFrame：

root
 |-- tickers: string (nullable = true)
 |-- name: string (nullable = true)
 |-- price: array (nullable = true)
 |    |-- element: map (containsNull = true)
 |    |    |-- key: string
 |    |    |-- value: map (valueContainsNull = true)
 |    |    |    |-- key: string
 |    |    |    |-- value: string (valueContainsNull = true)

我想将每个对象保存price到一个单独的 JSON 文件中，并使用相应的name字符串作为文件名保存每个文件。有没有办法在 Python 环境中实现它？

我找到的最相关的解决方案是将数据帧重新分区为数据帧中“行”数的分区，并使用.write.csv()（参见/sf/answers/3492341331/）。但这不符合我将“行”保存到具有不同文件名的单独文件的需要。

给出更多的上下文。我正在使用 spark 调用 API 并并行检索数据。spark数据框中的每一“行”都是基于 的唯一值的数据查询tickers。我的过程的最后一步是分别保存每个查询结果。如果有人有更好的方法来做到这一点，也会很感激。

非常感谢！

为了确保两个数据集的共置和共分区，它们必须在同一作业中由相同的键和分区数进行分区。

如果我加入这些数据集，结果加入的数据集会保留这个分区吗？

如果我然后在同一作业中使用相同的键和分区数对第三个数据集进行分区，这是否保证与加入的数据集的共同分区/托管？

我的数据如下所示：

+-------+-------+------+----------+
|book_id|user_id|rating|prediction|
+-------+-------+------+----------+
|    148|    588|     4|  3.953999|
|    148|  28767|     3| 2.5816362|
|    148|  41282|     3|  4.185532|
|    148|  18313|     4| 3.6297297|
|    148|  11272|     3| 3.0962112|
+-------+-------+------+----------+

我想通过四舍五入预测列中的值来创建一个新的列名“pred_class”。我运行这个代码：

results.withColumn('pred_class',round(results['prediction']))

它给了我这样的错误：

类型错误：类型列没有定义圆形方法

任何人都可以帮助我吗？谢谢！