我试图从字典中创建一个pysparkDataframe，然后对它执行一些查询。
这本词典可能有30万英镑key:value pairs)

example:  

  import pandas as pd
  import numpy as np
  import pyspark.sql.functions as F
  from pyspark.sql.types import IntegerType

  d = {0: [(0, 20.815), (1, 5.6802), (2, 31.208)], 1: [(0, 1.285), (1, 0.125)]} # each list  could have 500+ tuples.

  products_pd = pd.DataFrame.from_dict(d, orient='index')
  # this will create many columns , it is really awful!
  products_pd.columns = ['productid_'+str(x)+'_val' for x in range(3)] 
  # mark each column with a sequential number
  products_pd['row_num'] = np.arange(len(products_pd))
  products_df = spark.createDataFrame(products_pd.astype(str))
  products_df = products_df.withColumn('row_num', F.col('row_num').cast(IntegerType()))

我需要将所有这些名为“productid\u x\u val”的列聚合到一个数组中（每个元素作为一个元组（product\u id，val））作为一个新列，并删除所有这些“productid\u x\u val”列，因为可能有500多个列的名称如下。然后，我需要创建两个新列，在新数组的所有元素中使用最大的“val”及其对应的“product\u id”。
例如，
新的df应该是这样的：

row_num    prorducts_id_vals (one single column of array of tuples)  max_product_id  max_product_val. 
  0          [(0, 20.815), (1, 5.6802), (2, 31.208)]                    2               31.208         # because 31.208 is max 
  1          [(0, 1.285), (1, 0.125)]                                   0                1.285         # because  1.285 is max

我的代码效率不高，因为它将创建许多列（可以是500+），然后在pyspark dataframe中将它们聚合为一列。有没有可能不使用pandas就从给定的字典创建pysparkDataframe，并将新列作为数组？
如何选择最大值及其产品id来创建新列？
谢谢