I succeeded to do some experimental evaluation, and it seems I correctly understood the code:
A partition that consist of hive-buckets is read bucket-file by bucket-file, which leads to the loss of internal sorting.

Does anyone have an opinion about my alternative idea of reading from multiple bucket-files simultaneously to keep that ordering?

Regarding the followup questions:

1. I found the `collect_list()`function, which seems provide what I want. However, I fail to collect more than one column. Is there a way to do basically: .agg(collect_list("*")) ?

2. I worked around that problem by writing and reading the table within the same context/session, so that the ephemeral metastore doesn't loose it's content. However, in general a hive-metastore seems to be required for a production usage, since there is only an ephemeral- and a hive-catalog implementation available in 2.0.0.

I would highly appreciate some feedback to my thoughts and questions

Am 31.08.2016 um 14:45 schrieb Fridtjof Sander:
Hi Spark users,

I'm currently investigating spark's bucketing and partitioning capabilities and I have some questions:

Let  T be a table that is bucketed and sorted by T.id and partitioned by T.date. Before persisting, T has been repartitioned by T.id to get only one file per bucket.
I want to group by T.id over a subset of T.date's values.

It seems to me that the best execution plan in this scenario would be the following:
- Schedule one stage (no exchange) with as many tasks as we have bucket-ids, so that there is a mapping from each task to a bucket-id
- Each tasks opens all bucket-files belonging to "it's" bucket-id simultaneously, which is one per affected partition T.date
- Since the data inside the buckets are sorted, we can perform the second phase of "two-phase-multiway-merge-sort" to get our groups, which can be "pipelined" into the next operator

From what I understand after scanning through the code, however, it appears to me that each bucket-file is read completely before the record-iterator is advanced to the next bucket file (see FileScanRDD , same applies to Hive). So a groupBy would require to sort the partitions of the resulting RDD before the groups can be emitted, which results in a blocking operation.

Could anyone confirm that I'm assessing the situation correctly here, or correct me if not?

Followup questions:

1. Is there a way to get the "sql" groups into the RDD API, like the RDD groupBy would return them? I fail to formulate a job like this, because a query with groupBy, that misses an aggregation function, is invalid.
2. I haven't simply testet this, because I fail to load a table with the specified properties like above:
After writing a table like this:
.write().partitionBy("date").bucketBy(4, "id").sortBy("id").format("json").saveAsTable("table");
I fail to read it again, with the partitioning and bucketing being recognized.
Is a functioning Hive-Metastore required for this to work, or is there a workaround?

I hope someone can spare the time to help me out here.

All the best,
Fridtjof