Processamento de dados não estruturados com manipuladores de procedimento e UDF¶

Download da biblioteca PDFBox¶

Antes de começar a escrever a UDF, faça o download do arquivo JAR da biblioteca PDFBox se você ainda não tiver feito isso. Ele será uma dependência para o código do manipulador. Posteriormente, você fará o upload do arquivo JAR da biblioteca para um estágio.

Baixe a última versão lançada da biblioteca na página de download da biblioteca Apache PDFBox.

Criação de estágios¶

Crie estágios para manter as bibliotecas de dependência do código do manipulador e o arquivo de dados que o código do manipulador lerá.

Usando o código abaixo, você criará estágios internos separados para manter:

• Um arquivo JAR da biblioteca que é uma dependência para o seu manipulador. Você fará referência ao estágio e arquivo JAR da UDF.

• Um arquivo de dados que seu código de manipulador irá ler.

O código no exemplo a seguir usa o comando CREATE STAGE criar os estágios necessários.

-- Create an internal stage to store the JAR files.
CREATE OR REPLACE STAGE jars_stage;

-- Create an internal stage to store the data files. The stage includes a directory table.
CREATE OR REPLACE STAGE data_stage DIRECTORY=(ENABLE=TRUE) ENCRYPTION = (TYPE='SNOWFLAKE_SSE');

Como carregar a biblioteca necessária e o arquivo PDF a ser lido¶

Conclua as etapas a seguir para fazer upload do arquivo de dependência JAR (com o código da biblioteca que processa o PDF) e o arquivo de dados (o arquivo PDF que o código do manipulador processará).

Você pode usar o arquivo PDF de sua escolha neste exemplo.

1. Copie o arquivo JAR para o Apache PDFBox do diretório temporário local para o estágio que armazena os arquivos JAR:

   Linux/Mac:

   PUT file:///tmp/pdfbox-app-2.0.27.jar @jars_stage AUTO_COMPRESS=FALSE;
   

   Copy

   Windows:

   PUT file://C:\temp\pdfbox-app-2.0.27.jar @jars_stage AUTO_COMPRESS=FALSE;
   

   Copy

2. Copie o arquivo PDF do diretório temporário local para o estágio que armazena os arquivos de dados:

   Linux/Mac:

   PUT file:///tmp/myfile.pdf @data_stage AUTO_COMPRESS=FALSE;
   

   Copy

   Windows:

   PUT file://C:\temp\myfile.pdf @data_stage AUTO_COMPRESS=FALSE;
   

   Copy

Como criar e chamar a UDF¶

Conclua as etapas a seguir para criar uma UDF que lê e processa arquivos PDF.

1. Cole e execute o seguinte código para criar uma UDF.

   Esse manipulador de UDF analisa documentos PDF e recupera seu conteúdo. O manipulador usa a classe InputStream para ler o arquivo. Para saber mais sobre a leitura de arquivos com InputStream, consulte Como ler um arquivo especificado dinamicamente com InputStream.

   CREATE FUNCTION process_pdf_func(file STRING)
   RETURNS STRING
   LANGUAGE JAVA
   RUNTIME_VERSION = 11
   IMPORTS = ('@jars_stage/pdfbox-app-2.0.27.jar')
   HANDLER = 'PdfParser.readFile'
   AS
   $$
   import org.apache.pdfbox.pdmodel.PDDocument;
   import org.apache.pdfbox.text.PDFTextStripper;
   import org.apache.pdfbox.text.PDFTextStripperByArea;
   
   import java.io.File;
   import java.io.FileInputStream;
   import java.io.IOException;
   import java.io.InputStream;
   
   public class PdfParser {
   
       public static String readFile(InputStream stream) throws IOException {
           try (PDDocument document = PDDocument.load(stream)) {
   
               document.getClass();
   
               if (!document.isEncrypted()) {
   
                   PDFTextStripperByArea stripper = new PDFTextStripperByArea();
                   stripper.setSortByPosition(true);
   
                   PDFTextStripper tStripper = new PDFTextStripper();
   
                   String pdfFileInText = tStripper.getText(document);
                   return pdfFileInText;
               }
           }
           return null;
       }
   }
   $$;
   

   Copy

2. Atualize a tabela de diretórios para o estágio data_stage com o comando ALTER STAGE:

   ALTER STAGE data_stage REFRESH;
   

   Copy

3. Chame a UDF para ler o arquivo PDF preparado e extraia seu conteúdo.

   O código no exemplo a seguir chama a UDF, passando um URL com escopo para tornar o código resistente a ataques de injeção de arquivos. Sempre use um URL com escopo quando o chamador da função também não for seu proprietário. Você pode passar o argumento de URL como um URL com escopo ou outra forma quando o chamador da UDF é também proprietário dela.

   SELECT process_pdf_func(BUILD_SCOPED_FILE_URL('@data_stage', '/myfile.pdf'));
   

   Copy

Como criar e chamar o procedimento¶

Conclua as etapas a seguir para criar um procedimento que leia e processe arquivos PDF.

1. Cole e execute o seguinte código para criar um procedimento.

   O manipulador deste procedimento analisa documentos PDF e recupera seu conteúdo. O manipulador usa a classe SnowflakeFile para ler o arquivo. Para saber mais sobre a leitura de arquivos com SnowflakeFile, consulte Como ler um arquivo especificado dinamicamente com SnowflakeFile.

   CREATE PROCEDURE process_pdf_proc(file STRING)
   RETURNS STRING
   LANGUAGE JAVA
   RUNTIME_VERSION = 11
   IMPORTS = ('@jars_stage/pdfbox-app-2.0.28.jar')
   HANDLER = 'PdfParser.readFile'
   PACKAGES = ('com.snowflake:snowpark:latest')
   AS
   $$
   import org.apache.pdfbox.pdmodel.PDDocument;
   import org.apache.pdfbox.text.PDFTextStripper;
   import org.apache.pdfbox.text.PDFTextStripperByArea;
   import com.snowflake.snowpark_java.types.SnowflakeFile;
   import com.snowflake.snowpark_java.Session;
   
   import java.io.File;
   import java.io.FileInputStream;
   import java.io.IOException;
   import java.io.InputStream;
   
   public class PdfParser {
   
       public static String readFile(Session session, String fileURL) throws IOException {
           SnowflakeFile file = SnowflakeFile.newInstance(fileURL);
           try (PDDocument document = PDDocument.load(file.getInputStream())) {
   
               document.getClass();
   
               if (!document.isEncrypted()) {
   
                   PDFTextStripperByArea stripper = new PDFTextStripperByArea();
                   stripper.setSortByPosition(true);
   
                   PDFTextStripper tStripper = new PDFTextStripper();
   
                   String pdfFileInText = tStripper.getText(document);
                   return pdfFileInText;
               }
           }
   
           return null;
       }
   }
   $$;
   

   Copy

2. Atualize a tabela de diretórios para o estágio data_stage com o comando ALTER STAGE:

   ALTER STAGE data_stage REFRESH;
   

   Copy

3. Chame o procedimento para ler o arquivo PDF preparado e extraia seu conteúdo.

   O código no exemplo a seguir passa um URL com escopo apontando para o arquivo PDF no estágio que você criou.

   CALL process_pdf_proc(BUILD_SCOPED_FILE_URL('@data_stage', '/UsingThird-PartyPackages.pdf'));
   

   Copy

Como criar estágio de dados¶

Crie um estágio usando o comando CREATE STAGE:

A seguinte instrução SQL cria um estágio interno para armazenar os arquivos de dados para o exemplo:

-- Create an internal stage to store the data files. The stage includes a directory table.
CREATE OR REPLACE STAGE data_stage DIRECTORY=(ENABLE=TRUE) ENCRYPTION = (TYPE='SNOWFLAKE_SSE');

Como carregar o arquivo CSV para leitura¶

Copie o arquivo CSV do diretório temporário local para o estágio que armazena os arquivos de dados:

Linux/Mac:

PUT file:///tmp/sample.csv @data_stage AUTO_COMPRESS=FALSE;

Copy

Windows:

PUT file://C:\temp\sample.csv @data_stage AUTO_COMPRESS=FALSE;

Copy

Como criar e chamar a UDTF¶

Este exemplo extrai o conteúdo de um conjunto específico de arquivos CSV e retorna as linhas em uma tabela. Ao processar os dados do arquivo à medida que são lidos na origem, você pode evitar possíveis erros de falta de memória que podem surgir quando o arquivo é muito grande.

O código no exemplo do manipulador UDTF a seguir usa SnowflakeFile para gerar um InputStream de um URL de arquivo para ler um arquivo CSV. (Em um manipulador de UDTF Java, o processamento de linha começa quando Snowflake chama o método process que você implementa.) O código usa o fluxo ao construir uma instância de uma classe CsvStreamingReader definida no próprio manipulador.

A classe CsvStreamingReader lê o conteúdo do fluxo de arquivo CSV recebido linha por linha, fornecendo uma maneira para outro código recuperar cada linha como um registro onde vírgulas delimitam colunas. O método process retorna cada registro conforme ele é lido no fluxo.

Para obter mais informações sobre como escrever funções tabulares definidas pelo usuário (UDTFs) com um manipulador Java, consulte UDFs tabulares (UDTFs) de Java.

Complete os seguintes passos para criar a UDTF de Java e faça o upload dos arquivos necessários:

1. Crie uma UDTF de Java que usa a classe SnowflakeFile:

   CREATE OR REPLACE FUNCTION parse_csv(file STRING)
   RETURNS TABLE (col1 STRING, col2 STRING, col3 STRING )
   LANGUAGE JAVA
   HANDLER = 'CsvParser'
   AS
   $$
   import org.xml.sax.SAXException;
   
   import java.io.*;
   import java.util.ArrayList;
   import java.util.List;
   import java.util.stream.Stream;
   import com.snowflake.snowpark_java.types.SnowflakeFile;
   
   public class CsvParser {
   
     static class Record {
       public String col1;
       public String col2;
       public String col3;
   
       public Record(String col1_value, String col2_value, String col3_value)
       {
         col1 = col1_value;
         col2 = col2_value;
         col3 = col3_value;
       }
     }
   
     public static Class getOutputClass() {
       return Record.class;
     }
   
     static class CsvStreamingReader {
       private final BufferedReader csvReader;
   
       public CsvStreamingReader(InputStream is) {
         this.csvReader = new BufferedReader(new InputStreamReader(is));
       }
   
       public void close() {
         try {
           this.csvReader.close();
         } catch (IOException e) {
           e.printStackTrace();
         }
       }
   
       Record getNextRecord() {
         String csvRecord;
   
         try {
           if ((csvRecord = csvReader.readLine()) != null) {
             String[] columns = csvRecord.split(",", 3);
             return new Record(columns[0], columns[1], columns[2]);
           }
         } catch (IOException e) {
           throw new RuntimeException("Reading CSV failed.", e);
         } finally {
           // No more records, we can close the reader.
           close();
         }
   
         // Return null to indicate the end of the stream.
         return null;
       }
     }
   
     public Stream<Record> process(String file_url) throws IOException {
       SnowflakeFile file = SnowflakeFile.newInstance(file_url);
   
       CsvStreamingReader csvReader = new CsvStreamingReader(file.getInputStream());
       return Stream.generate(csvReader::getNextRecord);
     }
   }
   $$
   ;
   

   Copy

2. Atualize a tabela de diretório para o estágio data_stage:

   ALTER STAGE data_stage REFRESH;
   

   Copy

3. Chame a UDTF de Java para ler um ou mais arquivos CSV preparados e extrair o conteúdo em formato de tabela:

   O código no exemplo a seguir chama a UDF, passando um URL com escopo para reduzir o risco de ataques de injeção de arquivo. Sempre utilizou uma URL com escopo quando o chamador da função não é também seu proprietário. Você pode passar o argumento de URL como um URL com escopo ou outra forma suportada quando o chamador da UDF é também proprietário dela.

   -- Input a file URL.
   SELECT * FROM TABLE(PARSE_CSV(BUILD_SCOPED_FILE_URL(@data_stage, 'sample.csv')));
   

   Copy