You should call pivot on gruped data, so first you need to group by date and then pivot by p_category:

Copy>>> df.groupBy('date').pivot('p_category').agg(countDistinct('pid').alias('cnt')).show()
+----------+----+----------+                                                    
|      date|flat|ultra_thin|
+----------+----+----------+
|2016-09-30|   1|         1|
+----------+----+----------+

You can only do .pivot on objects having pivot attribute (method or property). You tried to do df.pivot, so it would only work if df had such attribute. You can inspect all the attributes of df (it's an object of pyspark.sql.DataFrame class) here. You see many attributes there, but none of them is called pivot. That's why you get an attribute error.

pivot is a method of pyspark.sql.GroupedData object. It means, in order to use it, you must somehow create pyspark.sql.GroupedData object from your pyspark.sql.DataFrame object. In your case, it's by using .groupBy():

df.groupBy("user_id").pivot("item_id")

This creates yet another pyspark.sql.GroupedData object. In order to make a dataframe out of it you would want to use one of the methods of GroupedData class. agg is the method that you need. Inside it, you will have to provide Spark's aggregation function which you will use for all the grouped elements (e.g. sum, first, etc.).

df.groupBy("user_id").pivot("item_id").agg(F.sum("watched_pct"))

Full example:

from pyspark.sql import functions as F

df = spark.createDataFrame(
    [(1, 1, '2021-05-11', 4250, 72),
     (1, 2, '2021-05-11', 80, 99),
     (2, 3, '2021-05-11', 1000, 80),
     (2, 4, '2021-05-11', 5000, 40)],
    ['user_id', 'item_id', 'last_watch_dt', 'total_dur', 'watched_pct'])

df = df.groupBy("user_id").pivot("item_id").agg(F.sum("watched_pct"))
df.show()
# +-------+----+----+----+----+
# |user_id|   1|   2|   3|   4|
# +-------+----+----+----+----+
# |      1|  72|  99|null|null|
# |      2|null|null|  80|  40|
# +-------+----+----+----+----+

If you want to replace nulls with 0, use fillna of pyspark.sql.DataFrame class.

df = df.fillna(0)
df.show()
# +-------+---+---+---+---+
# |user_id|  1|  2|  3|  4|
# +-------+---+---+---+---+
# |      1| 72| 99|  0|  0|
# |      2|  0|  0| 80| 40|
# +-------+---+---+---+---+

AttributeError: 'NoneType' object has no attribute 'unstack'

When you use inplace = True in set_index it modified the dataframe in place. It doesn't return anything(None). So you can't use unstack on None object.

inplace : boolean, default False

Modify the DataFrame in place (do not create a new object)

Use:

df1 = pdDF.set_index(['items_id','responseTime','label']).unstack('label')    
print(df1)

# Output:

id  responseTime    category_1  category_2 category_3 category_8
ABC  2018-06-24           [3]     [10]         [10]       NULL
DEF  2018-06-25           [7]     NULL         NULL       [10]
GHI  2018-06-28           NULL    NULL         [7]        NULL

Double check if there's a space in the column name. 'Survived ' vs 'Survived' It happens more often than you'd think especially with CSV data source.

After applying pivot you need to perform an aggregate, in this case the aggregate is first as the count metric has already been computed.

from pyspark.sql import functions as F

df = spark.createDataFrame([(123, 1, 1, ), 
                            (245, 1, 3), 
                            (123, 2, 5),], 
                           ("hashtag_id", "user_id", "count", ))

df.groupBy("user_id")\
  .pivot("hashtag_id")\
  .agg(F.first("count"))\
  .show()

Output

+-------+---+----+
|user_id|123| 245|
+-------+---+----+
|      1|  1|   3|
|      2|  5|null|
+-------+---+----+