pythongoogle-bigquerygoogle-cloud-storageavro
Load a avro file in BigQuery - Unexpected type for default value. Expected null, but found string: "null"
I need to transfer the result of this query to BigQuery, as you can see, I decode the data I get in Cloud Storage, I created an avro file to load it into a BigQuery table but I receive this error:
BadRequest Traceback (most recent call last)
<ipython-input-8-78860f4800c4> in <module>
110 bucket_name1 = 'gs://new_bucket/insert_transfer/*.avro'
111
--> 112 insert_bigquery_avro(bucket_name1, dataset1, tabela1)
<ipython-input-8-78860f4800c4> in insert_bigquery_avro(target_uri, dataset_id, table_id)
103 )
104 print('Starting job {}'.format(load_job.job_id))
--> 105 load_job.result()
106 print('Job finished.')
107
c:\users\me\appdata\local\programs\python\python37\lib\site-packages\google\cloud\bigquery\job.py in result(self, timeout)
695 self._begin()
696 # TODO: modify PollingFuture so it can pass a retry argument to done().
--> 697 return super(_AsyncJob, self).result(timeout=timeout)
698
699 def cancelled(self):
c:\users\me\appdata\local\programs\python\python37\lib\site-packages\google\api_core\future\polling.py in result(self, timeout)
125 # pylint: disable=raising-bad-type
126 # Pylint doesn't recognize that this is valid in this case.
--> 127 raise self._exception
128
129 return self._result
BadRequest: 400 Error while reading data, error message: The Apache Avro library failed to parse the header with the following error: Unexpected type for default value. Expected null, but found string: "null"
This is the script process:
import csv
import base64
import json
import io
import avro.schema
import avro.io
from avro.datafile import DataFileReader, DataFileWriter
import math
import os
import gcloud
from gcloud import storage
from google.cloud import bigquery
from oauth2client.client import GoogleCredentials
from datetime import datetime, timedelta
import numpy as np
try:
script_path = os.path.dirname(os.path.abspath(__file__)) + "/"
except:
script_path = "C:\\Users\\me\\Documents\\Keys\\key.json"
#Bigquery Credentials and settings
os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = script_path
folder = str((datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d'))
bucket_name = 'gs://new_bucket/table/*.csv'
dataset = 'dataset'
tabela = 'table'
schema = avro.schema.Parse(open("C:\\Users\\me\\schema_table.avsc", "rb").read())
writer = DataFileWriter(open("C:\\Users\\me\\table_register.avro", "wb"), avro.io.DatumWriter(), schema)
def insert_bigquery(target_uri, dataset_id, table_id):
bigquery_client = bigquery.Client()
dataset_ref = bigquery_client.dataset(dataset_id)
job_config = bigquery.LoadJobConfig()
job_config.schema = [
bigquery.SchemaField('id','STRING',mode='REQUIRED')
]
job_config.source_format = bigquery.SourceFormat.CSV
job_config.field_delimiter = ";"
uri = target_uri
load_job = bigquery_client.load_table_from_uri(
uri,
dataset_ref.table(table_id),
job_config=job_config
)
print('Starting job {}'.format(load_job.job_id))
load_job.result()
print('Job finished.')
#insert_bigquery(bucket_name, dataset, tabela)
def get_data_from_bigquery():
"""query bigquery to get data to import to PSQL"""
bq = bigquery.Client()
#Busca IDs
query = """SELECT id FROM dataset.base64_data"""
query_job = bq.query(query)
data = query_job.result()
rows = list(data)
return rows
a = get_data_from_bigquery()
length = len(a)
line_count = 0
for row in range(length):
bytes = base64.b64decode(str(a[row][0]))
bytes = bytes[5:]
buf = io.BytesIO(bytes)
decoder = avro.io.BinaryDecoder(buf)
rec_reader = avro.io.DatumReader(avro.schema.Parse(open("C:\\Users\\me\\schema_table.avsc").read()))
out=rec_reader.read(decoder)
writer.append(out)
writer.close()
def upload_blob(bucket_name, source_file_name, destination_blob_name):
storage_client = storage.Client()
bucket = storage_client.get_bucket(bucket_name)
blob = bucket.blob("insert_transfer/" + destination_blob_name)
blob.upload_from_filename(source_file_name)
print('File {} uploaded to {}'.format(
source_file_name,
destination_blob_name
))
upload_blob('new_bucket', 'C:\\Users\\me\\table_register.avro', 'table_register.avro')
def insert_bigquery_avro(target_uri, dataset_id, table_id):
bigquery_client = bigquery.Client()
dataset_ref = bigquery_client.dataset(dataset_id)
job_config = bigquery.LoadJobConfig()
job_config.autodetect = True
job_config.source_format = bigquery.SourceFormat.AVRO
time_partitioning = bigquery.table.TimePartitioning(type_=bigquery.TimePartitioningType.DAY, field="date")
job_config.time_partitioning = time_partitioning
uri = target_uri
load_job = bigquery_client.load_table_from_uri(
uri,
dataset_ref.table(table_id),
job_config=job_config
)
print('Starting job {}'.format(load_job.job_id))
load_job.result()
print('Job finished.')
dataset1 = 'dataset'
tabela1 = 'table'
bucket_name1 = 'gs://new_bucket/insert_transfer/*.avro'
insert_bigquery_avro(bucket_name1, dataset1, tabela1)
I receive a CSV file in Cloud Storage like this:
And this script decode the register like this:
I want to create a routine to put the decoded information into BigQuery.
The schema file:
{
"namespace": "transfers",
"type": "record",
"name": "Transfer",
"doc": "Represents the The transfer request",
"fields": [
{
"name": "id",
"type": "string",
"doc": "the transfer request id"
},
{
"name": "date",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
},
"doc": "the date where the transaction happend"
},
{
"name": "merchant",
"type": "string",
"doc": "the merchant who owns the payment"
},
{
"name": "amount",
"type": ["null", {
"type": "bytes",
"logicalType": "decimal",
"precision": 4,
"scale": 2
}],
"default": "null",
"doc": "the foreign amount for the payment"
},
{
"name": "status",
"type": {
"type": "enum",
"name": "transfer_status",
"symbols": [
"RECEIVED",
"WAITING_TRANSFER",
"ON_PROCESSING",
"EXECUTED",
"DENIED"
]
},
"default": "DENIED"
},
{
"name": "correlation_id",
"type": ["null", "string"],
"default": "null",
"doc": "the correlation id of the request"
},
{
"name": "transfer_period",
"type": ["null", "string"],
"default": "null",
"doc": "The transfer period spec"
},
{
"name": "payments",
"type": {
"type": "array",
"items": "string"
}
},
{
"name": "metadata",
"type": {
"type": "map",
"values": "string"
}
},
{
"name": "events",
"type": {
"type": "array",
"items": {
"name": "event",
"type": "record",
"fields": [
{
"name": "id",
"type": "string"
},
{
"name": "type",
"type": {
"type": "enum",
"name": "event_type",
"symbols": [
"REQUEST",
"VALIDATION",
"TRANSFER_SCHEDULE",
"TRANSFERENCE"
]
}
},
{
"name": "amount",
"type": ["null", {
"type": "bytes",
"logicalType": "decimal",
"precision": 4,
"scale": 2
}],
"doc": "the original currency amount",
"default": "null"
},
{
"name": "date",
"type": {
"type": "long",
"logicalType": "timestamp-millis"
},
"doc": "the moment where this request was received by the platform"
},
{
"name": "status",
"type": {
"type": "enum",
"name": "event_status",
"symbols": [
"SUCCESS",
"DENIED",
"ERROR",
"TIMEOUT",
"PENDING"
]
}
},
{
"name": "metadata",
"type": {
"type": "map",
"values": "string"
}
},
{
"name": "internal_metadata",
"type": {
"type": "map",
"values": "string"
}
},
{
"name": "error",
"type": {
"type": "record",
"name": "Error",
"fields": [
{
"name": "code",
"type": ["null", "string"],
"default": "null"
},
{
"name": "message",
"type": ["null", "string"],
"default": "null"
}
]
}
},
{
"name": "message",
"type": ["null", "string"],
"default": "null"
}
]
}
}
}
]
}
Solution
Try changing the "default" values from "null" to null.
- Python method chaining in functional programming style
- flask-jwt-extended: Fake Authorization Header during testing (pytest)
- For loop through the list unless empty?
- Polars make all groups the same size
- Is there a way to specify a default base-template for all templates in django?
- How to tackle time limit exceeded error in leetcode
- Is pd.get_dummies() updated in newer versions of Pandas making it default to Booleans (True/False) instead of (0/1)?
- What's the function like sum() but for multiplication? product()?
- How to type hint a dynamically-created dataclass
- Issue with pulling the data with EIA API with Python
- 403 Forbidden Error when scraping a site, user-agents already used and updated. Any ideas?
- Fullstack web-hosting services
- How to handle an AnalysisException on Spark SQL?
- Python requests is slow and takes very long to complete HTTP or HTTPS request
- Is there a way to modify an element in a Numpy array based on the value of other elements?
- Tkinter grid manager height/width nonconsistent
- Sql Alchemy Insert Statement failing to insert, but no error
- How can I create a Polars struct while eval-ing a list?
- Excel using win32com and python
- django: on pypy, psyco, unladen swallow or cpython, which one is the fastest?
- How convert a list into multiple columns and a dataframe?
- Name not defined in type annotation
- Static type checkers and Language Servers not recognizing attributes of objects that are subclasses
- How do I get multiple OID values in PySNMP?
- how to log a file in Django
- Is there a simple and efficient way to evaluate Elementary Symmetric Polynomials in Python?
- Iterating over two lists one after another
- Python -i flag for production
- What is PyCompilerFlags in Python C API?
- How to make Python check whether a variable is a number or letter