TypeError: Object of type Timestamp is not JSON serializable or TypeError: Object of type date is not JSON serializable

default

str

databricks_pyspark_step_launcher

queryArgs = f"""{{
            selector:{{
                repositoryName: "{repo_name}"
                repositoryLocationName: "{repo_loc}"
                pipelineName:"{pipeline_name}"}}
            runConfigData:{run_config}
            mode:"{mode_def}"
            }}"""
pipeline_execute = f"""mutation {{
                        launchPipelineExecution(executionParams:{queryArgs})
                        {{
                            __typename
                            ... on PipelineConfigValidationInvalid{{
                                errors{{
                                    message
                                }}
                            }}

                        }}
                        }}"""
print(pipeline_execute)
resp = json.loads(client.execute(pipeline_execute))

run_config = f"""{{
    num: 10,
    dict: {{hello:"Whats up"}},
    string: "Hello"}}"""

@solid
def hello():
    return 1


@solid(
    input_defs=[
        InputDefinition("name", str),
        InputDefinition("age", int),
    ]
)
def hello2(hello_: int, name: str, age: int):
    print(f"Hello {name}, {age+hello_}")


@pipeline
def pipe_2():
    x = hello()
    y = hello2(x, "Marcus", 20)


@repository
def deploy_docker_repository():
    return [pipe_2, my_schedule]

dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline pipe_2, received invalid type <class 'str'> for input "name" (at position 1) in solid invocation "hello2". Must pass the output from previous solid invocations or inputs to the composition function as inputs when invoking solids during composition.

grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with: status = StatusCode.UNAVAILABLE details = "failed to connect to all addresses" debug_error_string = "{"created":"@1630669409.194800714","description":"Failed to pick subchannel","file":"src/core/ext/filters/client_channel/client_channel.cc","file_line":3008,"referenced_errors":[{"created":"@1630669409.194797596","description":"failed to connect to all addresses","file":"src/core/ext/filters/client_channel/lb_policy/pick_first/pick_first.cc","file_line":397,"grpc_status":14}]}" >

grpcurl -plaintext <host>:4000 list

Failed to list services: server does not support the reflection API

Error loading repository location glue.py:dagster.core.errors.DagsterInvariantViolationError: Attempted to iterate over an InvokedSolidOutputHandle. This object represents the output "result" from the solid "load_partitions". Consider yielding multiple Outputs if you seek to pass different parts of this output to different solids.

@solid(
    description="Read from google sheet for data pipeline.",
    output_defs=[
        OutputDefinition(name="row_data"),
        OutputDefinition(name="gspread_object")
        ]
    )
def read_from_gsheet() -> List:
    print("Instantiate gspread sheet object and read a row..."
    row = sheet.row_values(2)
    yield Output(row, output_name="row_data")
    yield Output(sheet, output_name="gspread_object")

@solid(description="Write to cells in Gsheet")
def write_to_gsheet(sheet, cell: str, value: int):
    sheet.update(cell, value)

@solid(description="Read from google sheet and query data from NLP")
def query_NLP_data(context, row):
    received_row = row
    print("Sending request to query NLP graph and return metrics")
    some_bol_metrics = 1000
    <http://context.log.info|context.log.info>(f"Returned NLP metrics: {some_bol_metrics}")
    return some_bol_metrics

@pipeline
def gather_pipeline():
    row, sheet = read_from_gsheet()
    some_bol_metrics = query_NLP_data(row)
    write_to_gsheet(sheet, 'H2', some_bol_metrics)

/home/dazza/anaconda3/envs/versedai/lib/python3.8/site-packages/dagster/core/definitions/composition.py:65: UserWarning: While in @pipeline context 'gather_pipeline', received an uninvoked solid 'write_to_gsheet'.
  warnings.warn(warning_message.strip())
/home/dazza/anaconda3/envs/versedai/lib/python3.8/site-packages/dagster/core/workspace/context.py:504: UserWarning: Error loading repository location gather_pipeline.py:dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline gather_pipeline, received invalid type <class 'str'> for input "cell" (at position 1) in solid invocation "write_to_gsheet". Must pass the output from previous solid invocations or inputs to the composition function as inputs when invoking solids during composition.

gRPC

Nothing

dagster.core.errors.DagsterInvalidDefinitionError: In @pipeline sample_example_pipeline, could not resolve input based on position at index 0 for invocation sample_example_table. Use keyword args instead, available inputs are: ['input_start1', 'input_start2']

Nothing

dagster

dagit

dagster

dagit

dagit.exe

bin

--target=

dagit.exe