Cannot ingest data using snowpipe more than once

Question

I am using the sample program from the Snowflake document on using Python to ingest the data to the destination table.

So basically, I have to execute put command to load data to the internal stage and then run the Python program to notify the snowpipe to ingest the data to the table.

This is how I create the internal stage and pipe:

create or replace stage exampledb.dbschema.example_stage;

create or replace pipe exampledb.dbschema.example_pipe 
as copy into exampledb.dbschema.example_table 
from
    (
        select
            t.*
        from
            @exampledb.dbschema.example_stage t
    ) 
file_format = (TYPE = CSV) ON_ERROR = SKIP_FILE;

put command:

put file://E:\\example\\data\\a.csv @exampledb.dbschema.example_stage OVERWRITE = TRUE;

This is the sample program I use:

from logging import getLogger
from snowflake.ingest import SimpleIngestManager
from snowflake.ingest import StagedFile
from snowflake.ingest.utils.uris import DEFAULT_SCHEME
from datetime import timedelta
from requests import HTTPError
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.serialization import Encoding
from cryptography.hazmat.primitives.serialization import PrivateFormat
from cryptography.hazmat.primitives.serialization import NoEncryption
import time
import datetime
import os
import logging

logging.basicConfig(
        filename='/tmp/ingest.log',
        level=logging.DEBUG)
logger = getLogger(__name__)

# If you generated an encrypted private key, implement this method to return
# the passphrase for decrypting your private key.
def get_private_key_passphrase():
  return '<private_key_passphrase>'

with open("E:\\ssh\\rsa_key.p8", 'rb') as pem_in:
  pemlines = pem_in.read()
  private_key_obj = load_pem_private_key(pemlines,
  get_private_key_passphrase().encode(),
  default_backend())

private_key_text = private_key_obj.private_bytes(
  Encoding.PEM, PrivateFormat.PKCS8, NoEncryption()).decode('utf-8')
# Assume the public key has been registered in Snowflake:
# private key in PEM format

# List of files in the stage specified in the pipe definition
file_list=['a.csv.gz']
ingest_manager = SimpleIngestManager(account='<account_identifier>',
                                     host='<account_identifier>.snowflakecomputing.com',
                                     user='<user_login_name>',
                                     pipe='exampledb.dbschema.example_pipe',
                                     private_key=private_key_text)
# List of files, but wrapped into a class
staged_file_list = []
for file_name in file_list:
    staged_file_list.append(StagedFile(file_name, None))

try:
    resp = ingest_manager.ingest_files(staged_file_list)
except HTTPError as e:
    # HTTP error, may need to retry
    logger.error(e)
    exit(1)

# This means Snowflake has received file and will start loading
assert(resp['responseCode'] == 'SUCCESS')

# Needs to wait for a while to get result in history
while True:
    history_resp = ingest_manager.get_history()

    if len(history_resp['files']) > 0:
        print('Ingest Report:\n')
        print(history_resp)
        break
    else:
        # wait for 20 seconds
        time.sleep(20)

    hour = timedelta(hours=1)
    date = datetime.datetime.utcnow() - hour
    history_range_resp = ingest_manager.get_history_range(date.isoformat() + 'Z')

    print('\nHistory scan report: \n')
    print(history_range_resp)

After running the program, I just need to remove the file in the internal stage:

REMOVE @exampledb.dbschema.example_stage;

The code works as expected for the first time but when I truncate the data on that table and run the code again, the table on snowflake doesn't have any data in it.

Do I miss something here? How can I make this code can run multiple times?

Update:

I found that if I use a file with a different name each time I run, the data can load to the snowflake table.

So how can I run this code without changing the data filename?

Sergiu Sergiu · Accepted Answer · 2021-09-07T15:41:07

Snowflake uses file loading metadata to prevent reloading the same files (and duplicating data) in a table. Snowpipe prevents loading files with the same name even if they were later modified (i.e. have a different eTag).

The file loading metadata is associated with the pipe object rather than the table. As a result:

Staged files with the same name as files that were already loaded are ignored, even if they have been modified, e.g. if new rows were added or errors in the file were corrected.
Truncating the table using the TRUNCATE TABLE command does not delete the Snowpipe file loading metadata.

However, note that pipes only maintain the load history metadata for 14 days. Therefore:

Files modified and staged again within 14 days: Snowpipe ignores modified files that are staged again. To reload modified data files, it is currently necessary to recreate the pipe object using the CREATE OR REPLACE PIPE syntax.

Files modified and staged again after 14 days: Snowpipe loads the data again, potentially resulting in duplicate records in the target table.

For more information have a look here

Cannot ingest data using snowpipe more than once

1 Answers