相关疑难解决方法(0)

按字段将数据划分为预定义分区计数的最佳方法是什么？

我目前通过指定partionCount = 600来分区数据.发现计数600为我的数据集/集群设置提供了最佳查询性能.

val rawJson = sqlContext.read.json(filename).coalesce(600)
rawJson.write.parquet(filenameParquet)

现在我想通过列'eventName'对这些数据进行分区,但仍然保持计数600.数据当前有大约2000个唯一的eventNames,加上每个eventName中的行数不统一.大约10个eventNames有超过50%的数据导致数据偏斜.因此,如果我像下面那样进行分区,它的性能不是很好.写入比没有写入时间长5倍.

val rawJson = sqlContext.read.json(filename)
rawJson.write.partitionBy("eventName").parquet(filenameParquet)

为这些场景划分数据的好方法是什么？有没有办法按eventName进行分区,但是将其扩展到600个分区？

我的架构如下所示:

{  
  "eventName": "name1",
  "time": "2016-06-20T11:57:19.4941368-04:00",
  "data": {
    "type": "EventData",
    "dataDetails": {
      "name": "detailed1",
      "id": "1234",
...
...
    }
  }
} 

谢谢!

我读到具有相同分区器的 RDD 将位于同一位置。这对我很重要，因为我想加入几个未分区的大型 Hive 表。我的理论是，如果我可以将它们分区（通过字段调用 date_day）并位于同一位置，那么我将避免 shuffling 。

这是我试图为每张桌子做的事情：

def date_day_partitioner(key):
  return (key.date_day - datetime.date(2017,05,01)).days

df = sqlContext.sql("select * from hive.table")
rdd = df.rdd
rdd2 = rdd.partitionBy(100, date_day_partitioner)
df2 = sqlContext.createDataFrame(rdd2, df_log_entry.schema)

print df2.count()

不幸的是，我什至无法测试我关于协同定位和避免改组的理论，因为当我尝试 partitionBy 时出现以下错误：ValueError: too many values to unpack

Traceback (most recent call last):
  File "/tmp/zeppelin_pyspark-118755547579363441.py", line 346, in <module>
    raise Exception(traceback.format_exc())
Exception: Traceback (most recent call last):
  File "/tmp/zeppelin_pyspark-118755547579363441.py", line 339, in <module>
    exec(code)
  File "<stdin>", line 15, in <module>
  File "/usr/lib/spark/python/pyspark/sql/dataframe.py", line 380, in count …