创建Snowflake Python存储过程时遭遇SyntaxError语法错误

错误详情 #

执行创建存储过程的代码时，返回如下语法错误：

File "_udf_code.py", line 1
    CREATE PROCEDURE prc_industry_data_quality()
           ^^^^^^^^^
SyntaxError: invalid syntax
 in function PYTHON_WORKSHEET with handler main

问题代码 #

CREATE PROCEDURE prc_industry_data_quality()
  RETURNS TABLE
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.8'
  PACKAGES = ('snowflake-snowpark-python', 'pandas')
  HANDLER = 'main'
AS
$$
import pandas as pd
from snowflake.snowpark.functions import col
import pandas as pd
from datetime import datetime

database = "DEV_NDVM533_DB"
schema = "CENTRAL_MDM_020_STG"

def return_pd_dataframes(session):

     # SQL queries
    query1 = """
SELECT 
    POLICY_REFERENCE,
    SECTION_REFERENCE,
    CASE
        WHEN SOURCE_SYSTEM IS NULL THEN ORIGINATING_SOURCE_SYSTEM
        ELSE SOURCE_SYSTEM
    END AS SYSTEM,
    AUDIT_CREATE_DATE_TIME,
    INDUSTRY_CODE,
    INDUSTRY,
    INDUSTRY_CODE_QUALIFIER
FROM 
    test_sect_ext;
"""
    
    query2 = """
    SELECT 
        NAICS_CODE AS INDUSTRY_CODE,
        NAICS_TITLE,
        VALID_FROM,
        VALID_TO 
    FROM 
        CENTRAL_MDM.NAICS_6_DIGIT;
    """
    
    query3 = "SELECT * FROM stg_industry_data_quality;"

    query4 = "SELECT * FROM stg_industry_data_quality_log;"
    
    # Execute the queries and create pandas DataFrames
    query_df = session.sql(query1)
    query_df = query_df.to_pandas()
    
    naics_codes_df = session.sql(query2)
    naics_codes_df = naics_codes_df.to_pandas()
    
    prev_data_df = session.sql(query3)
    prev_data_df = prev_data_df.to_pandas()

    log_df = session.sql(query4)
    log_df = log_df.to_pandas()

    return query_df, naics_codes_df, prev_data_df, log_df


def preProcessData(data):
    subset = ['POLICY_REFERENCE', 'SYSTEM', 'INDUSTRY_CODE', 'INDUSTRY', 'INDUSTRY_CODE_QUALIFIER']

    data['POLICY_REFERENCE'] = data['POLICY_REFERENCE'].astype('str')
    data['AUDIT_CREATE_DATE_TIME'] = pd.to_datetime(data['AUDIT_CREATE_DATE_TIME'])
    data['SYSTEM'] = data['SYSTEM'].astype('str')
    data['INDUSTRY_CODE'] = data['INDUSTRY_CODE'].astype('str')
    data['INDUSTRY'] = data['INDUSTRY'].astype('str')
    data['INDUSTRY_CODE'] = data['INDUSTRY_CODE'].astype('str')
    data['INDUSTRY_CODE'] = data['INDUSTRY_CODE'].replace({'None': None})
    data['INDUSTRY_CODE_QUALIFIER'] = data['INDUSTRY_CODE_QUALIFIER'].astype('str')
    data['INDUSTRY_CODE_QUALIFIER'] = data['INDUSTRY_CODE_QUALIFIER'].replace({'None': None})
    data.drop_duplicates(subset= subset, keep="first", inplace=True)
    
    return data

# Evaluates if the value in a column is Null
def isNull(row, column): # For Industry, column = 'Industry_Code'
        if row[column] is None:
            return True
        else:
            return False

# Evaluates if the value in a column is found in the NAICS code lookup table - if so, returns True
def isInLookup(row, column, lookuptable, lookupcolumn):
    if row[column] in lookuptable[lookupcolumn].values:
         return True
    else:
        return False

# Evaluate if the value in a column conforms to validity rules (within an allowed date range)
def isValid(row):
    if row['IS_IN_LOOKUP'] == True:
        if row['AUDIT_CREATE_DATE_TIME'] >= row['VALID_FROM'] and row['AUDIT_CREATE_DATE_TIME'] <= row['VALID_TO']:
            return True
        else:
            return False
    else:
        return False


def addIsValidColumn(data, lookup_table):
    """Applies the isValid function to the main dataset by joining the data with the lookup table 
    to allow comparison of ['AuditCreateDateTime']with ['Valid From'] and ['Valid To'] columns"""

    # left join the lookup table onto the data
    result_data = data.copy()
    merged_df = data.merge(lookup_table, on='INDUSTRY_CODE', how='left')

    # Apply the is_valid function to each row and create the 'IsValid' column
    merged_df['IS_VALID'] = merged_df.apply(isValid, axis=1)

    # Create a new DataFrame that includes only 'PolicyReference' and 'IsValid' columns
    policy_validity = merged_df[['POLICY_REFERENCE', 'IS_VALID']]

    # Join the 'IsValid' information into the copy of the original data on 'PolicyReference' to not change the df's structure
    result_data = result_data.merge(policy_validity, on='POLICY_REFERENCE', how='left')

    return result_data

# Returns the reason for the exception
def Reason(row):
    if row['IS_NULL'] == True:
        return 'Data Missing'
    if row['IS_IN_LOOKUP'] == False:
        return 'Data Not In Lookup'
    if row['IS_VALID'] == False:
        return 'Invalid'
    else:
        return 'No Exception Found'
    

# Evaluates if a row (policy) in the data has been loaded before in the previous iteration. If so, returns True
def scannedBefore(row, prev_data):
    prev_policies = prev_data['POLICY_REFERENCE'].tolist()
    return row['POLICY_REFERENCE'] in prev_policies

def fixedReferences(prev_data, new_data):
    """Returns policies that were not fixed in the last run but are fixed in the current run"""

        # Filter for where 'Was Fixed'== 'False' in prev_data and == True in current data
    if not prev_data.empty:
        prev_data_filtered = prev_data[prev_data['WAS_FIXED'] == False]
        current_data_filtered = new_data[new_data['WAS_FIXED'] == True]
        # Find the common policy references between previous and current data
        fixed_references = set(prev_data_filtered['POLICY_REFERENCE']).intersection(set(current_data_filtered['POLICY_REFERENCE']))
    else:
        fixed_references = []
    return fixed_references

# Changed with Sean
def wasFixed(row, prev_data):
    """Evaluates if a row that was already in the previous iteration has been fixed."""
    
    prev_row = prev_data.loc[prev_data['POLICY_REFERENCE'] == row['POLICY_REFERENCE']]
    scanned_before = scannedBefore(row, prev_data)
    return scanned_before and prev_row['REASON'].iloc[0] != 'No Exception Found' and row['REASON'] == 'No Exception Found'


def ingestDate(row, prev_data):
    """Returns the current datetime if the row is not found in the previous iteration of data 
    i.e. it has just been ingested"""

    if 'INGEST_DATE' in row:
        if pd.isna(row['INGEST_DATE']):
            return datetime.now().date()
        else:
            return row['INGEST_DATE']
    else:
        return datetime.now().date()


def fixDate(row, fixed_references):
    """Returns the date that the 'Was_Fixed' column was recorded as being set to True"""
    if row['POLICY_REFERENCE'] in list(fixed_references) and 'FIX_DATE' in row and pd.isnull(row['FIX_DATE']):
        return datetime.now().date()
    elif 'FIX_DATE' in row and pd.notna(row['FIX_DATE']):
        return row['FIX_DATE']
    else:
        return None

def timeToFix(row):
    """Returns the time taken to fix an exception(in days)"""

    if row['FIX_DATE'] is not None:
        return int((row['FIX_DATE'] - row['INGEST_DATE']).days)
    else:
        return None


def updateLog(log_df, data):
    most_recent_date = data['INGEST_DATE'].max()

    if not log_df.empty:
        iteration = log_df['ITERATION'].max() + 1
    else:
        iteration = 1
    log_df.loc[iteration, 'ITERATION'] = iteration
    log_df.loc[iteration, 'REFRESH_DATE'] = datetime.now()
    log_df.loc[iteration, 'INVALID_COUNT'] = len(data[(data['IS_VALID'] == False) | (data['IS_VALID'] == None) | (data['IS_NULL'] == True) | (data['IS_IN_LOOKUP'] == False)])
    log_df.loc[iteration, 'RESOLVED_COUNT'] = len(data[data['WAS_FIXED'] == True])
    log_df.loc[iteration, 'NEW_EXCEPTIONS'] = ((data['INGEST_DATE'] == most_recent_date) & (data['REASON'].isin(['Data Missing', 'Data Not In Lookup', 'Invalid']))).sum()

    return log_df

def addAdditionalColumns(data, column, lookup_table, prev_data):
    """Applies the above functions to each row to create columns with logic behind them to be used in the data quality dashboard"""
    new_data = data.copy()
    
    new_data['IS_NULL'] = new_data.apply(lambda row: isNull(row, 'INDUSTRY_CODE'), axis=1) # working
    new_data['IS_IN_LOOKUP'] = new_data.apply(lambda row: isInLookup(row, 'INDUSTRY_CODE', lookup_table, 'INDUSTRY_CODE'), axis=1) # working
    new_data = addIsValidColumn(new_data, lookup_table) # working
    new_data['REASON'] = new_data.apply(lambda row: Reason(row), axis=1) # working
    new_data['INGEST_DATE'] = new_data.apply(lambda row: ingestDate(row, prev_data), axis=1) # working
    new_data['WAS_FIXED'] = new_data.apply(lambda row: wasFixed(row, prev_data), axis=1) # working
    if not prev_data.empty:
        fixed_references = fixedReferences(prev_data, new_data)
    else:
        fixed_references = []
    new_data['FIX_DATE'] = new_data.apply(lambda row: fixDate(row, fixed_references), axis=1) # working
    new_data['TIME_TO_FIX'] = new_data.apply(lambda row: timeToFix(row), axis=1) # working
        
    return new_data

def main(session: snowpark.Session):
    session.sql("USE DATABASE dev_ndvm533_db").collect()
    session.sql("USE SCHEMA central_mdm_020_stg").collect()
    query, lookup_table, prev_data, log = return_pd_dataframes(session)
    query = preProcessData(query)
    transformed_data = addAdditionalColumns(query, 'INDUSTRY_CODE', lookup_table, prev_data)
    log_df = log
    log_df = updateLog(log, transformed_data)

    # Convert transformed pandas DataFrame to Snowpark DataFrame
    snowpark_df = session.create_dataframe(transformed_data)
    snowpark_log = session.create_dataframe(log_df)
    
    # Write the Snowpark DataFrame to Snowflake tables
    snowpark_df.write.mode("overwrite").save_as_table("stg_industry_data_quality")
    snowpark_log.write.mode("append").save_as_table("stg_industry_data_quality_log")

    return snowpark_log.limit(10)

result = main(session)
$$;

解决方案 #

1. 切换执行环境：创建存储过程的语句是SQL语法，必须在Snowflake的SQL Worksheet中执行，不能在Python Worksheet中运行。Python Worksheet会将所有内容当作Python代码解析，因此会把CREATE PROCEDURE识别为无效Python语法。

2. 移除多余代码：Python代码块中的result = main(session)是多余的，Snowflake会通过指定的HANDLER = 'main'自动调用main函数，无需手动执行这一行。删除该行可以避免潜在的执行问题。

修改后的Python代码块末尾应去掉result = main(session)这一行。

内容的提问来源于stack exchange，提问作者fil_sql