why we need to uppercase columns? find error when dataframe like this

pd.DataFrame(['510365552', '564032331', '505153022', '547927024', '556154751', '569685159'])

TypeError: descriptor 'upper' for 'str' objects doesn't apply to a 'int' object
...

  File "....virtualenvs\dagster\lib\site-packages\dagster_gcp_pandas\bigquery\bigquery_pandas_type_handler.py", line 39, in handle_output
    with_uppercase_cols = obj.rename(str.upper, copy=False, axis="columns")

how to store df with json column or column with lists? gbq knows array ond json types. in

BigQueryPandasTypeHandler

in my case obj.dtypes returns ('items', dtype('O')),

context.add_output_metadata(
            {
                "row_count": obj.shape[0],
                "dataframe_columns": MetadataValue.table_schema(
                    TableSchema(
                        columns=[
                            TableColumn(name=name, type=str(dtype))
                            for name, dtype in obj.dtypes.iteritems()
                        ]
                    )
                ),
            }
        )

can i use sqlalchemy.types ?

json_columns = ['foo','baz']
data_type = dict.fromkeys(json_columns, types.JSON) if json_columns else {}
df.to_sql(
    table_name,
    connection,
    dtype=data_type,
    if_exists="append",
    index=False,
    chunksize=1000,
)

https://cloud.google.com/bigquery/docs/reference/standard-sql/data-types#array-type

debugging thread for @Benedikt Buchert

Hi, how can I configure the dataset and location to read from? It is an asset produced by dbt. I read https://docs.dagster.io/_apidocs/libraries/dagster-gcp-pandas#dagster_gcp_pandas.bigquery_pandas_io_manager Currently it is looking in

marts

but the correct dataset is prefixed by dbt

abc_marts

. Also it is looking inside US location, which is not where the dataset is located. I tried setting the default location to

"location": "europe-west3"

in my definitions which seems to have no effect. Is it possible to specify the dataset location and datset name per asset or asset group as well?

from dagster import asset, AssetIn
import pandas as pd

@asset(
    ins={"dim_ga4__users": AssetIn(
        key_prefix=["dbt_models", "marts"]
    )},
    group_name="marts"
)
def show_users_head(dim_ga4__users) -> pd.DataFrame:
    print(dim_ga4__users.head)
    return dim_ga4__users

Hi, I'm trying to make an orchestrator to send some sql queries to call some procedures. I understand that i have to use dagster_gcp.bq_op_for_queries(_sql_queries_), but i dont understand how to pass credentials (Expects a BQ client to be provisioned in resources as context.resources.bigquery) Is there some example I can use as a template. Thank you!

😛lanet-daggy: New for the BigQuery I/O manager in 0.17.21 😛lanet-daggy: 😒park: The BigQuery I/O manager now supports PySpark DataFrames 📚 New tutorial and reference guide goes over how to set up and configure the BigQuery I/O manager, storing and loading assets, working with partitions, and more! This release also includes a bugfix where

location

configuration was not being used when storing Pandas DataFrames Thanks again to everyone who has tried out the I/O manager and given feedback so far!

:blob-sun: New for the BigQuery I/O manager in 0.18.0 :blob-sun: When storing and fetching BigQuery tables as Pandas DataFrames, the I/O manager now uses the

google.bigquery.Client

methods

load_table_from_dataframe

and

query

instead of the

pandas_gbq

library. This shouldn’t cause any changes in functionality on your end, but do open a GitHub issue if you run into any problems

🌼 New for the BigQuery I/O manager in 0.18.2 🌼 This release highlight comes to you a few days late, but here’s what’s new in last week’s release. The BigQuery I/O manager supports self-dependent assets! If you have a partitioned asset that depends on a prior partition of itself, the BigQuery I/O manager will load that partition as a DataFrame. For the first partition in the series, an empty DataFrame will be returned

@asset(
    partitions_def=DailyPartitionsDefinition(start_date="2023-01-01"),
    ins={
        "self_dependent_asset": AssetIn(
            key=AssetKey(["self_dependent_asset"]),
            partition_mapping=TimeWindowPartitionMapping(start_offset=-1, end_offset=-1),
        ),
    },
    metadata={
        "partition_expr": "date",
    },
)
def self_dependent_asset(context, self_dependent_asset: pd.DataFrame) -> pd.DataFrame:
    date = pd.Timestamp(context.asset_partition_key_for_output())

    if self_dependent_asset.empty:
        return pd.DataFrame({"date": [date], "value": [1]})
    return pd.DataFrame({"date": [date], "value": [self_dependent_asset["value"][0] + 1]})

The BigQuery I/O manager now accepts

timeout

configuration. Currently, this configuration will only be used when working with Pandas DataFrames. It will set the number of seconds to wait for a request before using a retry.

👋

🌷 New for the BigQuery I/O manager + resource in 0.18.7 🌷 In this release, we have added new

BigQueryPandasIOManager

(api docs) and

BigQueryPySparkIOManager

(api docs) that follow the new Pythonic Config and Resources system. The existing

bigquery_pandas_io_manager

and

bigquery_pyspark_io_manager

are still and will continue to be supported. However, we think that the new Pythonic system has huge ergonomic benefits so we recommend you check it out! You can read more about the Pythonic Config and Resources system here , and expect to see even more resources (including updated guides) in the coming weeks! Additionally, the BigQuery resource now accepts authentication credentials via configuration (like the BigQuery I/O manager).

Hi! I would like to ask if I can use the Class-based resource mentioned in this Github page on

bigquery_resource

API from

dagster_gcp

?

🦚 New for the BigQuery I/O manager in 0.19.0 🦚 Happy 1.3 release everyone! In this release, we’ve updated the BigQuery tutorial and reference page to use the updated pythonic config and resources version of the BigQuery I/O manager.

Hi! I'm curious if there are thoughts on how to handle schema changes? The way the IO manager is currently configured for non partitioned tables it truncates the table explicitly which maintains the schema. the bigquery

load_table_from_dataframe

by default doesn't allow column addition. The 2 config options that bigquery seems to expose are: • setting the job WriteDisposition to

WRITE_TRUNCATE

(this replaces the whole table and is nice because it also allows incompatible changes like changing a column type, but doesn't fit in as well with the DBIOManager abstraction and would only work for non partitioned tables) • passing

ALLOW_FIELD_ADDITION

as a schemaUpdateOption I'm not sure the best way to expose these options but one or both would very helpful (we currently have a handwritten bq io manager that we've been using which uses `WRITE_TRUNCATE`ours doesn't support partitions though so excited to migrate!)

Hello! I am trying to gather metadata about dbt runs in bigquery. I am attempting to use the

runtime_metadata_fn

to gather this metadata about partitioned builds. The custom function I am using and the

load_assets_from_dbt_project

read as follows:

def custom_metadata_fn(context, node_info):
    table_name = node_info["name"]
    table_schema = node_info["schema"]
    n_rows = f"SELECT 1 from {table_schema}.{table_name} where date = {context.partition_key}"
    return {"n_rows": n_rows}

dbt_assets = load_assets_from_dbt_project(
    project_dir=DBT_PROJECT_PATH,
    profiles_dir=DBT_PROFILES,
    use_build_command=True,
    partitions_def=DailyPartitionsDefinition(start_date="2023-04-14"),
    partition_key_to_vars_fn=partition_key_to_dbt_vars,
    runtime_metadata_fn=custom_metadata_fn,
)

The question I have is how do I pass that

n_rows

query to either an io_manager or some other mechanism to actually query the

{table_schema}.{table_name}

table? do i just have to access the bigquery sdk? or is there a dagster api I can interface with? using:

dbt-bigquery==1.1.0
dagster==1.3.0
dagit==1.3.0
dagster-dbt==0.19.0

Hello! I am currently making a pipeline where I need do send message to Slack whenever a table in my

Bigquery

is updated(a row is added). I'm wondering if I could use

sensor

to monitor new changes to my table, and perhaps use

bigquery_resource

in an

asset

to retrieve data from Bigquery?

Hello! Question, Can I deploy dagster to GCP cloud run without postgresql?, I mean, just pack everything as a Docker container

and execute from cloud run?

without Postgresql and dagit?

Hey there! I am using the bigquery pandas io manager to periodically sync small tables to bigquery. However I am noticing that there are small periods of time where the table is empty. It seems like the operation is not atomic so there is a period of time when refreshing the table in which there is no data. Is there a way to make this atomic?

Hi, is there an example somewhere in the docs using the

BigQUeryResource

to load data incrementally into a table in BQ? I am thinking the static partitions will do but I'm not sure if that'll give me what I want. Essentially, i have data that comes at a regular frequency. For each time the data comes, the output of the asset is essentially the name of the data for that delivery period. I would want it to be appended to the table in BQ. Is this possible currently with the BigqueryResource?

😛otted_plant: New in dagster-gcp (BigQuery) in 0.19.3 😛otted_plant: A new

BigQueryResource

follows the new Pythonic Config and Resources system. The existing

bigquery_resource

will continue to be supported. This BigQueryResource works slightly differently than the

bigquery_resource

. Using the new

BigQueryResource

you create a

bigquery.Client

(bigquery docs) and use that to run queries

from dagster import asset 
from dagster_gcp import BigQueryResource 

@asset
def small_petals(bigquery: BigQueryResource):
    with bigquery.get_client() as client:
        return client.query(
            (
                'SELECT * FROM IRIS.IRIS_DATA WHERE "petal_length_cm" < 1 AND'
                ' "petal_width_cm" < 1'
            ),
        ).result()

I'm using the bigquery pandas IO manager to load data in partitions. I noticed that when the first partition is an empty dataframe, the table is created in BQ nut the data types are all wrong, which is expected because there's no data to use to infer the data types. Can we have something in the

handle_output

section of the pandas type handler file that skips the creation of tables if the data frame has 0 rows or some sort of a check to ensure that 0 rows data frames are not passed as output to the IO manager

Hello ! I might have missed something but I'm trying to have an asset that queries a table on BigQuery, transform data and then push on BigQuery To query I'm trying the brand new

BigQueryResource

(and it works like a charm 😛artydagster: ) but if I then try to add an io_manager linked to BigQuery on the same asset I have the following error:

Resource config error: gcp_credentials config for BigQuery resource cannot be used if GOOGLE_APPLICATION_CREDENTIALS environment variable is set.

My asset is declared this way:

@asset(io_manager_key="bigquery_io_manager")
def apiBatch(bigquery: BigQueryResource):

And my definition is:

defs = Definitions(
    assets = all_assets,
    resources={
        "bigquery": BigQueryResource(
                project = "PROJECT",
                location = "EU", 
                gcp_credentials = EnvVar("GCP_CREDS"),
        ),
        "bigquery_io_manager": BigQueryPandasIOManager(
            project="PROJECT",  
            location="EU",  
            dataset="mydataset",  
            gcp_credentials = EnvVar("GCP_CREDS"),
            timeout=15.0,  
        )
    },
)

I don't have this issue if I declare the io_manager_key to another asset; is it normal?

Hey! Is it possible to chose between

WRITE_TRUNCATE

and

WRITE_APPEND

with

BigQueryPandasIOManager

? Seems like it's always an overwrite (I haven't tried partitioned assets, it would overwrite only the partition?)

Hello! I would like to ask if it's possible to use

yield

instead of

return

to push data into

BigQuery

with

BigQueryPandasIOManager

? I have a loop that pushes data into

BigQuery

every iteration, so theoretically I would need to use

yield

so that the asset won't terminate. However, I would get this error when using

yield

any help would be very helpful!!

Hey everyone! I'm trying to create a partitioned table in BigQuery via

BigQueryPandasIOManager

. The partition in Dagster itself works (using

DailyPartitionsDefinition

), but the result in BigQuery is a regular table instead of one with the table type

Partitioned

. Is there a way to specify the table type directly via Dagster or is my only option to create the table manually in BigQuery and populate it via Dagster?

Hi there! Im trying to output the result of a software-defined asset to an existing BigQuery table. I use the

BigQueryPandasIOManager

. When the job is going to write to the table it fails because one of the columns in the dataframe is being translated to integer instead of string. This seems to happen because the column in the dataframe is None and dagster by default may infer that the type is an integer. This is the error I get:

google.api_core.exceptions.BadRequest: 400 Provided Schema does not match Table innate-empire-283902:enrichment.person_batch_enrichment_output. Field error_code has changed type from STRING to INTEGER

Is there a way to workaround this? How can I tell dagster that the output type of the column is a string?

Hey everybody! Currently, we pull data from a REST API via Python, store it via

BigQueryPandasIOManager

, and then model the data via DBT. Is there a way to skip the saving step and pass the data directly to DBT? This would allow us to omit saving the raw data and just store the final modeled data.

Hey all, beginner question here but is there a way to let dagster update the schema of output asset tables in BQ just based on code changes? I’m developing some models as assets and when I change the model in dagster code I currently have to delete the output table in BQ and then re-materialized the asset. Would be great if there’s a configuration I’m missing that can just overwrite the output table regardless.

We want to store newsletter data in BigQuery. Since the statistics change even after sending (open rate, click rate), we need to update the data retroactively. Currently, we have defined the data as a partitioned asset in Dagster, which we export to BQ via BigQueryPandasIOManager. Using the UI or the CLI (via dagster job backfill), backfilling the data works fine. However, we currently fail to schedule the job. build_schedule_from_partitioned_job would only fill the last partition, even per Freshness Policy and Auto Materialization we got nowhere. Is there a way to schedule the backfill of a partitioned asset in BigQuery? Maybe by triggering a CLI action via scheduled job?