Facing issue while querying from two tables using 'with' in sql

Vishal Anand 0
Viewed 43
spark_df = spark.sql("""
with api_users as 
    (
        select 
            distinct(user_id)
        from 
            act_post
        where 
            dt = '2022-08-15'
            and upper(api_source) = 'API'
        limit 100 
    )
select 
    user_id
from 
    (
    unified_user_interest_profile
    )
where user_id in api_users.user_id
limit 100 
""")

ERROR

Py4JJavaError: An error occurred while calling o75.sql.
: org.apache.spark.sql.catalyst.parser.ParseException: 
mismatched input 'from' expecting <EOF>(line 15, pos 0)

== SQL ==

with api_users as 
    (
        select 
            distinct(user_id)
        from 
            act_post
        where 
            dt = '2022-08-15'
            and upper(api_source) = 'API'
        limit 100 
    )
select 
    user_id
from 
^^^
    (
    unified_user_interest_profile
    )
where user_id in api_users.user_id
limit 100 
.
.
.
.
.

ParseException: "\nmismatched input 'from' expecting <EOF>(line 15, pos 0)\n\n== SQL ==\n\nwith api_users as \n    (\n        select \n            distinct(user_id)\n        from \n            act_post\n        where \n            dt = '2022-08-15'\n            and upper(api_source) = 'API'\n        limit 100 \n    )\nselect \n    user_id\nfrom \n^^^\n    (\n    unified_user_interest_profile\n    )\nwhere user_id in api_users.user_id\nlimit 100 \n"

I am trying to create a temporary table as api_users and using the user_id column to query all the records from another table unified_user_interest_profile where user_id is present in api_users.user_id

I am getting the above error and I couldn't understand what is the issue with it.

I am using this SQL command in pyspark to fetch the data.

mysql pyspark apache-spark sql
1 Answers

The conditions looks pretty straightforward, so you can just do an inner join based on user_id

from pyspark.sql import functions as F

api_users = (spark
    .table('act_post')
    .where((F.col('dt') == '2022-08-15') & (F.upper('api_source') == 'API'))
    .select('user_id')
    .distinct()
)

(spark
    .table('unified_user_interest_profile')
    .join(api_users, on=['user_id'], how='inner')
    .show()
)
pltc0
Related