Subir un archivo CSV a una tabla en GCP BigQuery

En una entrega pasada vimos como cargar datos desde un CSV a una tabla en BigQuery.

Esta vez crearemos un programa con Java y BigQuery. Para ello necesitamos tener un archivo con datos el cual llamaremos "DATOS.txt". El contenido del archivo será similar a esto:

20250412,34,450.0
20250412,34,432.0
20250412,122,500.0

Tendremos una tabla llamada ``tkdata`` con los siguientes campos:

fchinf DATE 
numregs INT64 
valor STRING 

Requisitos:

• Tener acceso a GCP BigQuery.
• Tener JDK 11 o más actual. 
• Tener Maven (el más actual).

¿Qué hará el programa?

1. Verificar si existe el archivo a subir ("DATOS.txt"). 
2. Obtener su contenido y guardarlo en una lista tipo String. 
3. Crear un objeto tipo StringBuilder a partir de la lista tipo String. 
4. Enviar el contenido del objeto StringBuilder a un nuevo archivo ("tkdata.csv") que se guardará en el mismo bucket del archivo original. 
5. Cargar el contenido del nuevo archivo a la tabla ``tkdata``.
6. Verificar que los datos hayan sido cargados a la tabla.

BigQueryCsvUploader.java

import com.google.cloud.bigquery.*;
import com.google.cloud.storage.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.StringReader;
import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class BigQueryCsvUploader {
    private static final Logger LOG = LoggerFactory.getLogger(BigQueryCsvUploader.class);
    private static final String NAME_FILE_ORIGINAL = "DATOS.csv";
    private static final String TABLE_NAME = "tkdata";
    private static final String DATASET = "mydataset";
    private static final String PROJECT = "myproject";
    private static final String BUCKET = "mybucket";
    private static final String NAME_NEW_FILE = "tkdata.csv";
    private static final long MAX_SIZE_BYTES = 300 * 1024 * 1024; // 300 MB

   
    public static class Tkdata {
        private Date fchinf;
        private long numregs;
        private String valor;

        public Date getFchinf() { return fchinf; }
        public void setFchinf(String fchinf) throws ParseException {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
            this.fchinf = sdf.parse(fchinf);
        }
        public long getNumregs() { return numregs; }
        public void setNumregs(String numregs) { this.numregs = Long.parseLong(numregs); }
        public String getValor() { return valor; }
        public void setValor(String valor) { this.valor = valor; }
    }

   
    public static boolean existFile(String bucketName, String fileName) {
        Storage storage = StorageOptions.getDefaultInstance().getService();
        Blob blob = storage.get(BlobId.of(bucketName, fileName));
        return blob != null && blob.exists();
    }

   
    public static boolean maxSize(String bucketName, 
     String fileName, long maxSizeBytes) {
        Storage storage = StorageOptions.getDefaultInstance().getService();
        Blob blob = storage.get(BlobId.of(bucketName, fileName));
        if (blob == null) {
            LOG.error("El archivo {} no existe en el bucket {}", fileName, bucketName);
            return false;
        }
        return blob.getSize() <= maxSizeBytes;
    }

    
    public static List<Tkdata> getListTkdata(String bucketName, String fileName) {
        List<Tkdata> listaTkdata = new ArrayList<>();
        Storage storage = StorageOptions.getDefaultInstance().getService();
        Blob blob = storage.get(BlobId.of(bucketName, fileName));
        if (blob == null) {
            LOG.error("Archivo{} no encontrado en el bucket {}", fileName, bucketName);
            return listaTkdata;
        }

        String content = new String(blob.getContent(), StandardCharsets.UTF_8);
        try (BufferedReader reader = new BufferedReader(new StringReader(content))) {
            String line;
            reader.readLine();
            while ((line = reader.readLine()) != null) {
                String[] parts = line.split(",", -1); 
                if (parts.length == 3) {
                    Tkdata obj = new Tkdata();
                    try {
                        obj.setFchinf(parts[0].trim());
                        obj.setNumregs(parts[1].trim());
                        obj.setValor(parts[2].trim());
                        listaTkdata.add(obj);
                    } catch (ParseException | NumberFormatException e) {
                        LOG.error("Error parseando linea: {}", line, e);
                    }
                } else {
                    LOG.warn("Linea malformada: {}", line);
                }
            }
        } catch (IOException e) {
            LOG.error("Error al leer el contenido", e);
        }
        return listaTkdata;
    }

    
    public static StringBuilder convertToStringBuilder(List<Tkdata> listaTkdata) {
        StringBuilder sb = new StringBuilder();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        for (Tkdata item : listaTkdata) {
            sb.append(sdf.format(item.getFchinf())).append(",");
            sb.append(item.getNumregs()).append(",");
            sb.append(item.getValor()).append("\n");
        }
        return sb;
    }

   
    public static boolean creaNuevoFileCSV(String bucketName, 
       String fileName, String content) {
        try {
            Storage storage = StorageOptions.getDefaultInstance().getService();
            BlobId blobId = BlobId.of(bucketName, fileName);
            BlobInfo blobInfo = BlobInfo.newBuilder(blobId).setContentType("text/csv").build();
            storage.create(blobInfo, content.getBytes(StandardCharsets.UTF_8));
            return true;
        } catch (Exception e) {
            LOG.error("Error al subir archivo {} al bucket {}", fileName, bucketName, e);
            return false;
        }
    }

    
    public static boolean loadCSVToTableBigQuery(String projectId, 
     String datasetId, String bucketName, String sourceFileName, String tableName) {
        try {
            BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
            TableId tableId = TableId.of(projectId, datasetId, tableName);

            
            Schema schema = Schema.of(
                Field.of("fchinf", StandardSQLTypeName.DATE),
                Field.of("numregs", StandardSQLTypeName.INT64),
                Field.of("valor", StandardSQLTypeName.STRING)
            );

           
            JobConfiguration jobConfig = LoadJobConfiguration.newBuilder(
                tableId,
                String.format("gs://%s/%s", bucketName, sourceFileName),
                FormatOptions.csv()
            )
                .setSchema(schema)
                .setSkipLeadingRows(0) 
                .setWriteDisposition(JobInfo.WriteDisposition.WRITE_APPEND) 
                .build();

            
            Job job = bigquery.create(JobInfo.of(jobConfig));
            job = job.waitFor();
            if (job.isDone() && job.getStatus().getError() == null) {
                LOG.info("CSV {} successfully loaded into BigQuery table {}.{}", 
                 sourceFileName, datasetId, tableName);
                return true;
            } else {
                LOG.error("Error cargando CSV a BigQuery: {}", job.getStatus().getError());
                return false;
            }
        } catch (Exception e) {
            LOG.error("Error en el proceso de carga", e);
            return false;
        }
    }

  
    public static boolean validateTableData(String projectId, 
      String datasetId, String tableName, List<Tkdata> expectedData) {
        try {
            BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
            String query = String.format("SELECT fchinf, numregs, valor FROM %s.%s.%s", 
                        projectId, datasetId, tableName);
            QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(query).build();

           
            TableResult result = bigquery.query(queryConfig);
            long rowCount = result.getTotalRows();

            if (rowCount == 0) {
                LOG.error("No hay datos en la tabla {}.{}", datasetId, tableName);
                return false;
            }

           
            if (expectedData != null && rowCount != expectedData.size()) {
                LOG.warn("Error en conteo de datos {}, found {}", expectedData.size(), rowCount);
                return false;
            }

            
            LOG.info("Sample data from {}.{} (first 5 rows):", datasetId, tableName);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
            int maxRowsToLog = 5;
            int rowIndex = 0;
            for (FieldValueList row : result.iterateAll()) {
                if (rowIndex >= maxRowsToLog) break;
                String fchinf = row.get("fchinf").getStringValue(); 
                long numregs = row.get("numregs").getLongValue();
                String valor = row.get("valor").getStringValue();
                LOG.info("Row {}: fchinf={}, numregs={}, valor={}", rowIndex + 1, fchinf, numregs, valor);
                rowIndex++;
            }

            LOG.info("Validacion exitosa: {} rows found in table {}.{}", 
          rowCount, datasetId, tableName);
            return true;
        } catch (Exception e) {
            LOG.error("Error validando datos en la tabla {}.{}", datasetId, tableName, e);
            return false;
        }
    }

    public static void main(String[] args) {
     
        if (!existFile(BUCKET, NAME_FILE_ORIGINAL)) {
            LOG.error("Archivo{} no existe en el buckett {}.", NAME_FILE_ORIGINAL, BUCKET);
            return;
        }

        
        if (!maxSize(BUCKET, NAME_FILE_ORIGINAL, MAX_SIZE_BYTES)) {
            LOG.error("El archivo {} excede el tamaño: 300MB.", NAME_FILE_ORIGINAL);
            return;
        }

        
        List<Tkdata> listaTkdata = getListTkdata(BUCKET, NAME_FILE_ORIGINAL);
        if (listaTkdata.isEmpty()) {
            LOG.error("No hay datos para leer {}. ", NAME_FILE_ORIGINAL);
            return;
        }

        StringBuilder sb = convertToStringBuilder(listaTkdata);

        
        if (creaNuevoFileCSV(BUCKET, NAME_NEW_FILE, sb.toString())) {
            LOG.info("Archivo {} cargado al bucket {}", NAME_NEW_FILE, BUCKET);
            LOG.info("Carga de datos {} a la tabla: {}.{}", NAME_NEW_FILE, DATASET, TABLE_NAME);
            if (loadCSVToTableBigQuery(PROJECT, DATASET, BUCKET, NAME_NEW_FILE, TABLE_NAME)) {
                if (validateTableData(PROJECT, DATASET, TABLE_NAME, listaTkdata)) {
                    LOG.info("Validacion correcta {}.{}", DATASET, TABLE_NAME);
                } else {
                    LOG.error("Validacion fallida {}.{}", DATASET, TABLE_NAME);
                }
            } else {
                LOG.error("Fallo al cargar los datos a la tabla de BigQuery  {}.{}", DATASET, TABLE_NAME);
            }
        } else {
            LOG.error("Fallo al cargar el archivo {} al bucket {}", NAME_NEW_FILE, BUCKET);
        }
    }
}

Es importante hacer notar que los tipos de datos en el archivo deben concordar a los datos de la tabla. En caso contrario, no cargarán.

También es importante tener estas dependencias en el pom.xml

<dependencies>
    <dependency>
        <groupId>com.google.cloud</groupId>
        <artifactId>google-cloud-storage</artifactId>
        <version>2.44.0</version>
    </dependency>
    <dependency>
        <groupId>com.google.cloud</groupId>
        <artifactId>google-cloud-bigquery</artifactId>
        <version>2.44.0</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>2.0.13</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>2.0.13</version>
    </dependency>
</dependencies>

De preferncia usar Eclipse IDE para generar un .jar o ejecutarlo directamente.

Si la ejecución es correcta podremos ver los datos cargados en la tabla. Continuaremos más sobre BigQuery en próximas entregas.

Enlaces:

https://cloud.google.com/bigquery/docs/loading-data-cloud-storage-csv

GCP: Crear un archivo CSV a partir de una consulta en BigQuery

En esta ocasión veremos como crear un archivo CSV a partir del resultado de una consulta en GCP BigQuery.

¿Qué haremos?

1. Crearemos una tabla temporal. 
2. Consultaremos una tabla de la cual queremos los datos.
3. Con los datos obtenidos crearemos un archivo CSV.

¿Qué es una tabla temporal en BigQuery?

Es una tabla creada temporalmente. Nos servirá como pivote para obtener ciertos datos de una consulta.

¿Cómo exportamos datos en BigQuery?

Usaremos la siguiente sentencia para exportar datos:

EXPORT DATA

El código es el siguiente:

CREATE OR REPLACE PROCEDURE `myproject.mydataset.create_file_csv`(input_fecha STRING)
BEGIN
  -- Crear una tabla temporal con los datos de la consulta
  CREATE TEMP TABLE temp_table AS
  SELECT * FROM `myproject.mydataset.Informe`
  WHERE fecha = CAST(input_fecha AS DATE);

  -- Exportar la tabla temporal a un archivo CSV en GCS, ya que BigQuery no soporta TXT directamente
  EXPORT DATA OPTIONS(
    uri='gs://your_bucket/path/to/file*.csv',
    format='CSV',
    overwrite=true,
    header=true
  ) AS
  SELECT * FROM temp_table;

END;

Para invocar el SP:

CALL `myproject.mydataset.create_file_csv`('2025-01-23');

Tendremos que ir a nuestro Bucket para comprobar que el archivo se ha creado.

En próximas entregas continuaremos con este tema.

Enlaces:

https://cloud.google.com/bigquery/docs/exporting-data

GCP BigQuery: cargar un CSV a una tabla y crear un respaldo

GCP BigQuery nos permite cargar datos de un fichero (con formato CSV, TXT, etc.) a tablas en nuestros datasets.

En ésta ocasión veremos como validar la existencia de un archivo CSV en un bucket (contenedor), cargarlo a una tabla temporal e insertarlo a una determinada tabla. Además veremos cómo crear un respaldo de ese archivo CSV a otro bucket.

BigQueryTransfer.java

import com.google.cloud.bigquery.*;
import com.google.cloud.storage.*;
import com.google.cloud.bigquerystorage.v1.*;
import com.google.protobuf.Descriptors;
import com.google.protobuf.Message;
import com.google.protobuf.util.JsonFormat;
import java.io.*;
import java.nio.channels.Channels;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;

public class BigQueryTransfer {

    private static final String BUCKET_NAME = "your-bucket-name";
    private static final String BACKUP_BUCKET_NAME = "your-backup-bucket-name";
    private static final String DATASET_NAME = "mydataset";
    private static final String TABLE_NAME = "transferencia";
    private static final String PROJECT_ID = "myproject";
    private static final String CSV_FILE_NAME = "datos.csv";
    private static final String RESULT_CSV_NAME = "resultados.csv";

    public static void main(String[] args) throws Exception {
        Storage storage = StorageOptions.getDefaultInstance().getService();
        BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();

        // 1. Verificar la existencia del archivo CSV
        Blob blob = storage.get(BUCKET_NAME, CSV_FILE_NAME);
        if (blob == null) {
            System.out.println("El archivo " + CSV_FILE_NAME + " no existe en el bucket.");
            return;
        }

        // 2. Leer el contenido del archivo CSV
        List<String> listaContenido = new ArrayList<>();
        try (BufferedReader reader = new BufferedReader(new InputStreamReader(Channels.newInputStream(blob.reader())))) {
            String line;
            while ((line = reader.readLine()) != null) {
                listaContenido.add(line);
            }
        }

        // 3. Crear la lista de Transferencia
        List<Transferencia> listaTransferencia = new ArrayList<>();
        for (String line : listaContenido) {
            String[] parts = line.split(",");
            if (parts.length >= 3) {
                Transferencia transferencia = new Transferencia(parts[0], parts[1], parts[2]);
                listaTransferencia.add(transferencia);
            }
        }

        // 4. Guardar en BigQuery usando BigQuery Storage Write API
        writeToBigQueryStorageApi(listaTransferencia);

        // 5. Crear "resultados.csv" con datos de la tabla
        String query = "SELECT * FROM `" + PROJECT_ID + "." + DATASET_NAME + "." + TABLE_NAME + "`";
        TableResult result = bigQuery.query(QueryJobConfiguration.newBuilder(query).build());

        File resultFile = new File(RESULT_CSV_NAME);
        try (BufferedWriter writer = new BufferedWriter(new FileWriter(resultFile))) {
            for (FieldValueList row : result.iterateAll()) {
                writer.write(String.join(",",
                        row.get(0).getStringValue(),
                        row.get(1).getStringValue(),
                        row.get(2).getStringValue()));
                writer.newLine();
            }
        }

        // Subir "resultados.csv" al bucket
        BlobId resultBlobId = BlobId.of(BUCKET_NAME, RESULT_CSV_NAME);
        BlobInfo resultBlobInfo = BlobInfo.newBuilder(resultBlobId).build();
        storage.create(resultBlobInfo, Files.readAllBytes(resultFile.toPath()));

        // 6. Mover "datos.csv" al bucket de respaldo
        BlobId sourceBlobId = BlobId.of(BUCKET_NAME, CSV_FILE_NAME);
        BlobId backupBlobId = BlobId.of(BACKUP_BUCKET_NAME, CSV_FILE_NAME);
        storage.copy(Storage.CopyRequest.of(sourceBlobId, backupBlobId));
        storage.delete(sourceBlobId);

        System.out.println("Proceso completado con éxito.");
    }

    private static void writeToBigQueryStorageApi(List<Transferencia> listaTransferencia) throws Exception {
        try (BigQueryWriteClient client = BigQueryWriteClient.create()) {
            String tablePath = String.format("projects/%s/datasets/%s/tables/%s", PROJECT_ID, DATASET_NAME, TABLE_NAME);

            WriteStream writeStream = WriteStream.newBuilder().setType(WriteStream.Type.COMMITTED).build();
            WriteStream createdStream = client.createWriteStream(CreateWriteStreamRequest.newBuilder().setParent(tablePath).setWriteStream(writeStream).build());

            ProtoSchema protoSchema = client.getWriteStream(GetWriteStreamRequest.newBuilder().setName(createdStream.getName()).build()).getTableSchema();
            Descriptors.Descriptor descriptor = TableSchemaToDescriptor.parse(protoSchema.getProtoDescriptor());

            for (Transferencia transferencia : listaTransferencia) {
                Message.Builder messageBuilder = DynamicMessage.newBuilder(descriptor);
                messageBuilder.setField(descriptor.findFieldByName("fecha"), transferencia.fecha);
                messageBuilder.setField(descriptor.findFieldByName("clave"), transferencia.clave);
                messageBuilder.setField(descriptor.findFieldByName("cuenta"), transferencia.cuenta);

                client.appendRows(AppendRowsRequest.newBuilder()
                        .setWriteStream(createdStream.getName())
                        .addRows(ByteString.copyFrom(messageBuilder.build().toByteArray()))
                        .build());
            }

            client.finalizeWriteStream(FinalizeWriteStreamRequest.newBuilder().setName(createdStream.getName()).build());
        }
    }

    static class Transferencia {
        String fecha;
        String clave;
        String cuenta;

        Transferencia(String fecha, String clave, String cuenta) {
            this.fecha = fecha;
            this.clave = clave;
            this.cuenta = cuenta;
        }
    }
}

Con este programa:

1. Verificamos la existencia del archivo "datos.csv". 
2. Leemos su contenido y lo almacenamos en una lista tipo String. 
3. Cargamos el contenido del archivo "datos.csv" a una tabla temporal. 
4. Creamos una lista de tipo "Transferencia". 
5. Guardamos los datos en la tabla "transferencia" usando BigQuery Storage Write API. 
6. Creamos un archivo en el bucket llamado "resultados.csv" a partir de una consulta a la tabla "transferencia". 
7. Creamos un respaldo de "datos.csv" a otro bucket.

Enlaces:

https://cloud.google.com/storage/docs/buckets
https://medium.com/@bravnic/bigquery-storage-write-api-at-scale-7affcc2d7a93
https://cloud.google.com/bigquery/docs/write-api
https://www.googlecloudcommunity.com/gc/Data-Analytics/Using-Google-Bigquery-Storage-Write-API-with-high-concurrency/m-p/662308