Title: Use python variables aggregating Spark dataframe
Tags: dataframe;apache-spark;datetime;pyspark;aggregation-framework
Question: I have a dataframe and a list of variables which is called customerids:
customerid
createdate
personid
birthday
genders
lastupdated
locale
parentid
profiles
providers
1028598965607080
2022-06-20 15:03:
2bfe51b1
2022-06-20 15:03:
00000000-0000-000
74c302ef
ba337d36
1020304050607099
2022-06-20 15:03:
04919240
1991-09-08
2022-06-20 15:03:
00000000-0000-000
2ee40b26
71024633
8423413884965465
2022-06-20 15:03:
3a97d280
2022-06-20 15:03:
00000000-0000-000
29d3a7b0
77973479
8948423132187895
2022-06-20 15:03:
4d6122b9
1991-12-23
2022-06-20 15:03:
00000000-0000-000
39f0ec37
c2774f60
1028598965607080
2022-06-20 15:03:
6e55e9f8
1995-09-15
2022-06-20 15:03:
00000000-0000-000
01d9ba4e
cf6b749a
1028598965607080
2022-06-20 15:03:
be76de55
1965-08-05
2022-06-20 15:03:
00000000-0000-000
896a97e6
d441f25f
8798432154654555
2022-06-20 15:03:
c9118c73
2022-06-20 15:03:
00000000-0000-000
01598fd4
648d5915
1231897897421258
2022-06-20 15:03:
e4893ee2
1995-04-12
2022-06-20 15:03:
00000000-0000-000
cf31c1a3
f0290168
List of variables:
customerids = [1028598965607080, 1020304050607099, 8423413884965465, 8948423132187895, 8798432154654555, 1231897897421258]
Other variable:
date = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S') - timedelta(days=1)
Basically, I need to do the select below in the dataframe, taking the variables of date and customerids to be able to generate a new dataframe with the column count:
SELECT count(1) FROM dataframe WHERE createdate <'{date} 21:00:00' and customerid = '{customerids}
How can I do this?
Expected output:
customerid
1028598965607080
1020304050607099
8423413884965465
8948423132187895
8798432154654555
1231897897421258

Here is the accepted answer: Dataframe:
from datetime import datetime, timedelta
from pyspark.sql import functions as F
df = spark.createDataFrame(
[(1028598965607080, '2022-06-20 15:03:01'),
(1020304050607099, '2022-06-20 15:03:01'),
(8423413884965465, '2022-06-20 15:03:01'),
(8948423132187895, '2022-06-20 15:03:01'),
(1028598965607080, '2022-06-20 15:03:01'),
(1028598965607080, '2022-06-20 15:03:01'),
(8798432154654555, '2022-06-20 15:03:01'),
(1231897897421258, '2022-06-20 15:03:01')],
['customerid', 'createdate'])
Your date variable has another variable inside (date_str). So I created it myself.
date_str = '2022-07-01 12:58:58'
customerids = [1028598965607080, 1020304050607099, 8423413884965465, 8948423132187895, 8798432154654555, 1231897897421258]
date = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S') - timedelta(days=1)
ts = f&quot;{date.year}, {date.month}, {date.day}, 21, 0, 0&quot;
df = (df.filter(F.col('customerid').isin(customerids) &amp;
(F.col('createdate') < F.expr(f&quot;make_timestamp({ts})&quot;)))
.groupBy('customerid')
.count()
df.show()
# +----------------+-----+
# |      customerid|count|
# +----------------+-----+
# |1028598965607080|    3|
# |1020304050607099|    1|
# |8423413884965465|    1|
# |8948423132187895|    1|
# |8798432154654555|    1|
# |1231897897421258|    1|
# +----------------+-----+


{'source': 'StackExchangeDataset', 'subset': 'stackoverflow.com', 'metadata': {'AnswerCount': 1, 'CommentCount': 0, 'ContentLicense': 'CC BY-SA 4.0', 'CreationDate': '2022:00:07/01/22', 'FavoriteCount': None, 'PostTypeId': 1, 'Tags': '<dataframe><apache-spark><datetime><pyspark><aggregation-framework>', 'Title': 'Use python variables aggregating Spark dataframe', 'ViewCount': 49, 'id': 72832841, 'user_AboutMe': "<p>Atuo como analista de BI, DBM e engenheira de dados desde 2019.\nSou graduada em Gestão da Tecnologia da Informação pela UNICID (2020) e atualmente estou cursando Banco de Dados (BI, Big Data e Data Science) na instituição FIAP.</p>\n<p>Possuo experiência com alguns bancos de dados relacionais sendo eles:</p>\n<ul>\n<li>SQL Server</li>\n<li>Teradata</li>\n<li>Oracle</li>\n<li>Amazon Redshift</li>\n</ul>\n<p>Já atuei com a ferramenta de ETL Integration Service (SSIS) da Microsoft desenvolvendo novos projetos e realizando sustentação dos mesmos e jobs já existentes.</p>\n<p>Atualmente trabalho com Engenharia de Dados aplicando meus conhecimentos com Python, Pyspark, bibilioteca Pandas, Amazon Redshift (DW) e orquestrador de dados Airflow.</p>\n<p>Alguns dos meus reconhecimentos acadêmicos são:</p>\n<ul>\n<li>Primeiro lugar no Enterprise Challenge (FIAP/2020)</li>\n<li>Primeiro lugar no Enterprise ODS (FIAP/2020)</li>\n</ul>\n<p>English:\nI have been working as a BI analyst, DBM and data engineer since 2019.\nI have a degree in Information Technology Management from UNICID (2020) and I am currently studying Database (BI, Big Data and Data Science) at the FIAP institution.</p>\n<p>I have experience with some relational databases, namely:</p>\n<ul>\n<li>SQL Server</li>\n<li>Teradata</li>\n<li>Oracle</li>\n<li>Amazon Redshift</li>\n</ul>\n<p>I've worked with Microsoft's ETL Integration Service (SSIS) tool developing new projects and supporting them and existing jobs.</p>\n<p>Currently I work with Data Engineering applying my knowledge with Python, Pyspark, Pandas library, Amazon Redshift (DW) and Airflow data orchestrator.</p>\n<p>Some of my academic recognitions are:</p>\n<ul>\n<li>First place in the Enterprise Challenge (FIAP/2020)</li>\n<li>First place in the Enterprise ODS (FIAP/2020)</li>\n</ul>\n<p>Français:\nJe travaille en tant qu'analyste BI, DBM et ingénieur de données depuis 2019.\nJ'ai un diplôme en gestion des technologies de l'information de l'UNICID (2020) et j'étudie actuellement la base de données (BI, Big Data et Data Science) à l'institution FIAP.</p>\n<p>J'ai de l'expérience avec certaines bases de données relationnelles, à savoir :</p>\n<ul>\n<li>SQL Server</li>\n<li>Teradata</li>\n<li>Oracle</li>\n<li>Amazon Redshift</li>\n</ul>\n<p>J'ai travaillé avec l'outil ETL Integration Service (SSIS) de Microsoft pour développer de nouveaux projets et les soutenir ainsi que les emplois existants.</p>\n<p>Actuellement, je travaille avec Data Engineering en appliquant mes connaissances avec Python, Pyspark, la bibliothèque Pandas, Amazon Redshift (DW) et l'orchestrateur de données Airflow.</p>\n<p>Certaines de mes reconnaissances académiques sont:</p>\n<ul>\n<li>Première place au Challenge Entreprise (FIAP/2020)</li>\n<li>Première place à l'Enterprise ODS (FIAP/2020)</li>\n</ul>\n", 'user_AccountId': 19624883.0, 'user_DisplayName': 'Caroline Leite', 'user_Reputation': 87.0, 'user_Views': 43.0, 'user_id': 18177021.0}}