Process unstructured data with UDF and procedure handlers¶

Este tópico fornece exemplos de leitura e processamento de dados não estruturados em arquivos preparados com código de manipulador escrito para o seguinte:

Você também pode ler um arquivo com manipuladores escritos em outras linguagens:

Python:

Scala:

Nota

Para tornar seu código resistente a ataques de injeção de arquivos, use sempre uma URL com escopo ao passar a localização de um arquivo para uma UDF, especialmente quando o chamador da função não for também seu proprietário. Você pode criar uma URL com escopo em SQL usando a função interna BUILD_SCOPED_FILE_URL. Para obter mais informações sobre o que o BUILD_SCOPED_FILE_URL faz, consulte Introdução aos dados não estruturados.

Como processar um PDF com uma UDF e procedimento¶

Os exemplos nesta seção processam arquivos não estruturados preparados usando o código do manipulador Java – primeiro com uma UDF, depois com um procedimento. Ambos os manipuladores extraem o conteúdo de um determinado arquivo PDF usando a bilioteca Apache PDFBox.

O código do manipulador é muito semelhante entre UDF e procedimento. Eles diferem em como lêem os dados recebidos do arquivo PDF.

Na UDF, o manipulador lê o arquivo usando um InputStream Java.
No procedimento, o manipulador lê o arquivo usando um SnowflakeFile Snowflake.

Os exemplos usam código manipulador em linha (ao contrário de compilado em um JAR), o que significa que você não precisa compilar, empacotar e fazer upload do código do manipulador para um estágio. Para obter mais informações sobre a diferença entre manipuladores em linha e em estágios, consulte Como manter o código do manipulador em linha ou em um estágio.

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.

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;
Windows:
PUT file://C:\temp\pdfbox-app-2.0.27.jar @jars_stage AUTO_COMPRESS=FALSE;
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;
Windows:
PUT file://C:\temp\myfile.pdf @data_stage AUTO_COMPRESS=FALSE;

Como criar e chamar a UDF¶

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

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;
    }
}
$$;

Atualize a tabela de diretórios para o estágio data_stage com o comando ALTER STAGE:
```
ALTER STAGE data_stage REFRESH;
```
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'));
```

Como criar e chamar o procedimento¶

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

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;
    }
}
$$;

Atualize a tabela de diretórios para o estágio data_stage com o comando ALTER STAGE:
```
ALTER STAGE data_stage REFRESH;
```
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'));
```

Processamento de um CSV com uma UDTF¶

O exemplo nesta seção extrai e retorna dados de arquivos preparados usando UDTFs de Java.

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;

Windows:

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

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:

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);
  }
}
$$
;

Atualize a tabela de diretório para o estágio data_stage:
```
ALTER STAGE data_stage REFRESH;
```
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')));