Как получить все продукты с хотя бы X обзорами в SQL?

Question

Я бы хотел найти все строки, в которых product_id имеет как минимум 3 отзыва (review_id).

Так что для следующего примера:

product_id, review_id
1,   1
1,   2
1,   3
2,   4
3,   5
4,   6
4,   7
4,   8

Должен вернуться:

product_id, review_id
1,   1
1,   2
1,   3
4,   6
4,   7
4,   8

Это запросы, которые я пробовал:

query1 = '''
        SELECT first_value(customer_id), first_value(review_id), first_value(product_id)
        FROM df
        WHERE product_category='Toys' OR product_category='Beauty'
        GROUP BY product_id
        HAVING COUNT(*) >= 5
'''

Выше приведен только первый результат, но когда я не использую функцию first_value, яполучить ошибку "expression 'df.review_id' is neither present in the group by, nor is it an aggregate function.

query1 = '''
        SELECT customer_id, review_id, product_id
        FROM df
        WHERE product_category='Toys' OR product_category='Beauty'
        GROUP BY product_id
        HAVING COUNT(*) >= 5
'''

Answer 1

Я хочу вернуть все результаты, если у продукта есть хотя бы 3 отзыва

О, боже, это такой фантастический вариант использования для оконной агрегации (которая часто является лучшей альтернативой groupBy, а затем join).

scala> input.show
+----------+---------+
|product_id|review_id|
+----------+---------+
|         1|        1|
|         1|        2|
|         1|        3|
|         2|        4|
|         3|        5|
|         4|        6|
|         4|        7|
|         4|        8|
+----------+---------+

import org.apache.spark.sql.expressions.Window
val productIds = Window.partitionBy("product_id")
val solution = input
  .withColumn("count", count('*) over productIds)
  .filter('count >= 3) // <-- that's the gist of the solution
  .select('product_id, 'review_id)
scala> solution.show
+----------+---------+
|product_id|review_id|
+----------+---------+
|         1|        1|
|         1|        2|
|         1|        3|
|         4|        6|
|         4|        7|
|         4|        8|
+----------+---------+

---

Я оставляю сравнение производительности оконного и простого агрегирования (groupBy + join) как домашнее упражнение:)

Answer 2

Использование Spark scala sql

val df = Seq((1,1),(1,2),(1,3),(2,4),(3,5),(4,6),(4,7),(4,8)).toDF("product_id", "review_id")
df.createOrReplaceTempView("review")
spark.sql(
  """ with t1( select product_id, review_id , count(*) over(partition by product_id) c1 from review)
    select product_id, review_id  from t1 where c1 >=3
  """).show(false)

Результаты:

+----------+---------+
|product_id|review_id|
+----------+---------+
|1         |1        |
|1         |2        |
|1         |3        |
|4         |6        |
|4         |7        |
|4         |8        |
+----------+---------+

Использование функций df для получения одинаковых результатов

import org.apache.spark.sql.expressions.Window
val df = Seq((1,1),(1,2),(1,3),(2,4),(3,5),(4,6),(4,7),(4,8)).toDF("product_id", "review_id")
df.withColumn("cn",count('product_id).over(Window.partitionBy('product_id))).filter(" cn>=3 ").drop("cn").show(false)

Answer 3

это будет работать:

CREATE TABLE d061_Table1(product_id int, review_id int);    
//do inserts


select * from d061_Table1 where product_id IN( select distinct product_id  from 
(select product_id,count(product_id) from d061_Table1
group by product_id
having count(product_id)>=3));

вывод:

1   1
1   2
1   3
4   6
4   7
4   8

Answer 4

Используя PySpark SQL, вы можете сделать

from pyspark.sql import functions as func
#Get num reviews per product
counts = df.groupBy(df.product_id).agg(func.countDistinct(df.review_id).alias('num_reviews'))
#Filter for num reviews >= 3
gt_3_reviews = counts.filter(counts.num_reviews >= 3).select(counts.product_id)
#Join it with the original dataframe and select the required columns
res = df.join(counts, counts.product_id == df.product_id).select(df.product_id,df.review_id)
res.show()

или использовать оконные функции

from pyspark.sql import functions as func
from pyspark.sql import Window 
#Select distinct rows from columns needed
dist_df = df.select(df.product_id,df.review_id).distinct()
#Define the window
w = Window.partitionBy(dist_df.product_id)
#Count number of reviews per product with the previously defined window
dist_df = dist_df.withColumn('num_reviews',func.count('*').over(w))
#Filter criteria
res = dist_df.filter(dist_df.num_reviews >= 3).select(dist_df.product_id,dist_df.review_id)
res.show()

Answer 5

Найдите все ваши самые популярные пункты в виде отдельного запроса и присоедините его, чтобы получить оставшиеся данные

SELECT df.* FROM
  df
  INNER JOIN
  (
    SELECT product_id
    FROM df
    WHERE product_category='Toys' OR product_category='Beauty'
    GROUP BY product_id
    HAVING COUNT(*) >= 5
  ) interesting
  on df.product_id = interesting.product_id

Это общий шаблон, который вам нужно использовать для группировки / подсчета, а затем для получения дополнительных данных.о продуктах, которые входят в группу.Вы не можете добавить больше столбцов к запросу, который выполняет группировку, потому что это разбивает группы на меньшие числа, опускаясь ниже порога.Вместо этого вы подсчитываете вхождения только идентификатора, ограничивает его только теми, у кого есть какое-то значение, а затем используете этот список интересных идентификаторов, чтобы извлечь остальную информацию для этих идентификаторов, присоединяя ее к таблице, которая имеетвсе данные

Базы данных, которые поддерживают аналитические запросы, могут делать это одним ударом, но приведенный выше шаблон запроса работает с базами данных, которые не поддерживают аналитику, поэтому я склонен использовать его.

БД, поддерживающий аналитику, можно запросить следующим образом:

  SELECT * FROM
  (
    SELECT *, COUNT(*) OVER(partition by product_id) numrev
    FROM df
    WHERE product_category='Toys' OR product_category='Beauty'
  ) interesting
  WHERE numrev >= 5

COUNT (*) OVER () по сути делает то же самое;база данных будет подсчитывать каждое вхождение продукта и представлять счет для продукта в каждой строке, тогда предложение where ограничивает только строки, количество которых превышает 4.Представьте, что в данном примере группировка по подзапросу скрыта в фоновом режиме (раздел является операцией группировки), и подразумевается объединение (поскольку подсчитанный идентификатор продукта размещается в каждой строке рядом с реальным идентификатором продукта дляк которому это относится)