https://linen.dev logo
Join Slack
Powered by
Hi all, first off want to say thanks to getting th...
# troubleshooting
j
Hi all, first off want to say thanks to getting the target-snowflake up to date with the Meltano SDK. I have included the new 
batch_size_rows
command in my meltano file, and I can see the command when I run 
meltano invoke target-snowflake --about
. However it seems that things aren't working quite as I expect it to - I have a (small) table of 940K rows and it is still chunking the 
.json
files into 10,000 entry increments rather than the amount I specified. 🧵 I am wondering if I am missing anything else configuration wise that I should be doing? I am running 
target-snowflake
(Meltano Variant) in conjunction with 
tap-mssql
(Buzzcut Norman) which is also on the Meltano SDK
👀 1
One of the JSON files still chunking into 10K rows
Interesting, maybe I am having incorrect assumptions of the expected end result of the data being pulled. Re-running I am seeing that the json.gz files have a (target-snowflake-table_name-GUID-#.json.gz) which goes up to 15 then starts over. I thought that the JSON files would have increments of rows based on what I tell it to, so each json would've had 150 K rows.
b
You are correct. Looking at the code for 
target-snowflake
it is using the 
JSONLinesBatcher.get_batches
. I think you can give it a batch config to change that maybe?
Copy code
batch_config:
  batch_size: 150000
j
As such?
Copy code
plugins:
  loaders:
  - name: target-snowflake
    variant: meltanolabs
    pip_url: meltanolabs-target-snowflake
    batch_config:
      batch_size: 150000
    config:
      add_record_metadata: false # Can enable if we want more metadata
      #account: See .ENV
      #database: TS See .ENV 
      #user: See .ENV
      #role: See .ENV
      #warehouse: See .ENV
      #password: See .ENV
      default_target_schema: Raw # ${MELTANO_EXTRACT__LOAD_SCHEMA} # Meltano chooses the schema based on the `name` of the extractor
      hard_delete: false
      batch_size_rows: 150000
👀 1
that didnt work, maybe i need to embed it under 
config
b
I think like this.
Copy code
plugins:
  loaders:
  - name: target-snowflake
    variant: meltanolabs
    pip_url: meltanolabs-target-snowflake
    config:
      add_record_metadata: false # Can enable if we want more metadata
      #account: See .ENV
      #database: TS See .ENV 
      #user: See .ENV
      #role: See .ENV
      #warehouse: See .ENV
      #password: See .ENV
      default_target_schema: Raw # ${MELTANO_EXTRACT__LOAD_SCHEMA} # Meltano chooses the schema based on the `name` of the extractor
      hard_delete: false
      batch_size_rows: 150000
      batch_config:
        batch_size: 150000
❤️ 1
is it working ?
j
One second; it's now having a bad time with that 
batch_config
being embedded, something about a pipe closing error. When I removed the 
batch_config
setting it started working again
😞 1
Copy code
024-03-27T20:04:53.442684Z [info     ] 2024-03-27 16:04:53,442 | INFO     | target-snowflake     | Target sink for 'Static-RegionCode' is full. Current size is '150000'. Draining... cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.443684Z [info     ] Traceback (most recent call last): cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.444867Z [info     ]   File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 196, in _run_module_as_main cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.444867Z [info     ]     return _run_code(code, main_globals, None, cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.445867Z [info     ]   File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\runpy.py", line 86, in _run_code cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.445867Z [info     ]     exec(code, run_globals)    cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.446868Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\Scripts\target-snowflake.exe\__main__.py", line 7, in <module> cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.446868Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\click\core.py", line 1157, in __call__ cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr
 string_id=target-snowflake
2024-03-27T20:04:53.447867Z [info     ]     return self.main(*args, **kwargs) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.447867Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\click\core.py", line 1078, in main cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.448870Z [info     ]     rv = self.invoke(ctx)      cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.448870Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\plugin_base.py", line 80, in invoke cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.449868Z [info     ]     return super().invoke(ctx) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.449868Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\click\core.py", line 1434, in invoke cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.450869Z [info     ]     return ctx.invoke(self.callback, **ctx.params) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.450869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\click\core.py", line 783, in invoke cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.451869Z [info     ]     return __callback(*args, **kwargs) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.451869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\target_base.py", line 567, in invoke cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.451869Z [info     ]     target.listen(file_input)  cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.452869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\io_base.py", line 35, in listen cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.452869Z [info     ]     self._process_lines(file_input) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.453869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\target_base.py", line 307, in _process_lines cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.453869Z [info     ]     counter = super()._process_lines(file_input) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.453869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\io_base.py", line 94, in _process_lines cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.454869Z [info     ]     self._process_record_message(line_dict) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.454869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\target_base.py", line 371, in _process_record_message cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.454869Z [info     ]     self.drain_one(sink)       cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.455868Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\target_base.py", line 512, in drain_one cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.455868Z [info     ]     sink.process_batch(draining_status) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.456868Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\sinks\sql.py", line 262, in process_batch cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.456868Z [info     ]     self.bulk_insert_records(  cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.456868Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\target_snowflake\sinks.py", line 147, in bulk_insert_records cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.457873Z [info     ]     batch_config=self.batch_config, cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.457873Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\target_snowflake\sinks.py", line 168, in batch_config cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.458869Z [info     ]     return BatchConfig.from_dict(raw) cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.458869Z [info     ]   File "C:\Git\elt-pipeline\.meltano\loaders\target-snowflake\venv\lib\site-packages\singer_sdk\helpers\_batch.py", line 255, in from_dict cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.458869Z [info     ]     return cls(**data)         cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.459869Z [info     ] TypeError: BatchConfig.__init__() missing 2 required positional arguments: 'encoding' and 'storage' cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.498527Z [info     ] 2024-03-27 16:04:53,498 | INFO     | snowflake.connector.connection | closed cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr string_id=target-snowflake
2024-03-27T20:04:53.499528Z [info     ] 2024-03-27 16:04:53,498 | INFO     | snowflake.connector.connection | No async queries seem to be running, deleting session cmd_type=elb consumer=True name=target-snowflake producer=False stdio=stderr
 string_id=target-snowflake
2024-03-27T20:04:53.672422Z [error    ] [WinError 109] The pipe has been ended
Copy code
2024-03-27T20:04:53.672422Z [error    ] [WinError 109] The pipe has been ended
╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\log │
│ ging\output_logger.py:207 in redirect_logging                                                    │
│                                                                                                  │
│   204 │   │   │   *ignore_errors,                                                                │
│   205 │   │   )                                                                                  │
│   206 │   │   try:                                                                               │
│ ❱ 207 │   │   │   yield                                                                          │
│   208 │   │   except ignored_errors:  # noqa: WPS329                                             │
│   209 │   │   │   raise                                                                          │
│   210 │   │   except RunnerError as err:                                                         │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\blo │
│ ck\extract_load.py:462 in run                                                                    │
│                                                                                                  │
│   459 │   │   │   # TODO: legacy `meltano elt` style logging should be deprecated                │
│   460 │   │   │   legacy_log_handler = self.output_logger.out("meltano", logger)                 │
│   461 │   │   │   with legacy_log_handler.redirect_logging():                                    │
│ ❱ 462 │   │   │   │   await self.run_with_job()                                                  │
│   463 │   │   │   │   return                                                                     │
│   464 │   │   else:                                                                              │
│   465 │   │   │   logger.warning(                                                                │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\blo │
│ ck\extract_load.py:494 in run_with_job                                                           │
│                                                                                                  │
│   491 │   │                                                                                      │
│   492 │   │   with closing(self.context.session) as session:                                     │
│   493 │   │   │   async with job.run(session):                                                   │
│ ❱ 494 │   │   │   │   await self.execute()                                                       │
│   495 │                                                                                          │
│   496 │   async def terminate(self, graceful: bool = False) -> None:                             │
│   497 │   │   """Terminate an in flight ExtractLoad execution, potentially disruptive.           │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\blo │
│ ck\extract_load.py:454 in execute                                                                │
│                                                                                                  │
│   451 │   │   async with self._start_blocks():                                                   │
│   452 │   │   │   await self._link_io()                                                          │
│   453 │   │   │   manager = ELBExecutionManager(self)                                            │
│ ❱ 454 │   │   │   await manager.run()                                                            │
│   455 │                                                                                          │
│   456 │   async def run(self) -> None:                                                           │
│   457 │   │   """Run the ELT task."""                                                            │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\blo │
│ ck\extract_load.py:659 in run                                                                    │
│                                                                                                  │
│   656 │   │   for IO to complete as appropriate. Expect a RunnerError to be raised if            │
│   657 │   │   any of the blocks exit with a non 0 exit code.                                     │
│   658 │   │   """                                                                                │
│ ❱ 659 │   │   await self._wait_for_process_completion(self.elb.head)                             │
│   660 │   │   _check_exit_codes(                                                                 │
│   661 │   │   │   self._producer_code,                                                           │
│   662 │   │   │   self._consumer_code,                                                           │
│ ───────────────────────────────────────────────────────────────────╯
`
Copy code
C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\blo │
│ ck\extract_load.py:732 in _wait_for_process_completion                                           │
│                                                                                                  │
│   729 │   │   │   │   │   line_length_limit=self.line_length_limit,                              │
│   730 │   │   │   │   │   stream_buffer_size=self.stream_buffer_size,                            │
│   731 │   │   │   │   )                                                                          │
│ ❱ 732 │   │   │   raise output_futures_failed.exception()  # noqa: RSE102                        │
│   733 │   │   else:                                                                              │
│   734 │   │   │   # If all the output handlers completed without raising an                      │
│   735 │   │   │   # exception, we still need to wait for all the underlying block                │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\log │
│ ging\utils.py:233 in capture_subprocess_output                                                   │
│                                                                                                  │
│   230 │   │   │   continue                                                                       │
│   231 │   │                                                                                      │
│   232 │   │   for writer in line_writers:                                                        │
│ ❱ 233 │   │   │   if not await _write_line_writer(writer, line):                                 │
│   234 │   │   │   │   # If the destination stream is closed, we can stop capturing output.       │
│   235 │   │   │   │   return                                                                     │
│   236                                                                                            │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\pipx\pipx\venvs\meltano\lib\site-packages\meltano\core\log │
│ ging\utils.py:201 in _write_line_writer                                                          │
│                                                                                                  │
│   198 │   │   │   writer.write(line)                                                             │
│   199 │   │   │   await writer.drain()                                                           │
│   200 │   │   except (BrokenPipeError, ConnectionResetError):                                    │
│ ❱ 201 │   │   │   await writer.wait_closed()                                                     │
│   202 │   │   │   return False                                                                   │
│   203 │   else:                                                                                  │
│   204 │   │   writer.writeline(line.decode())                                                    │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\streams.py:343 in    │
│ wait_closed                                                                                      │
│                                                                                                  │
│   340 │   │   return self._transport.is_closing()                                                │
│   341 │                                                                                          │
│   342 │   async def wait_closed(self):                                                           │
│ ❱ 343 │   │   await self._protocol._get_close_waiter(self)                                       │
│   344 │                                                                                          │
│   345 │   def get_extra_info(self, name, default=None):                                          │
│   346 │   │   return self._transport.get_extra_info(name, default)                               │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\proactor_events.py:3 │
│ 85 in _loop_writing                                                                              │
│                                                                                                  │
│   382 │   │   │   self._write_fut = None                                                         │
│   383 │   │   │   self._pending_write = 0                                                        │
│   384 │   │   │   if f:                                                                          │
│ ❱ 385 │   │   │   │   f.result()                                                                 │
│   386 │   │   │   if data is None:                                                               │
│   387 │   │   │   │   data = self._buffer                                                        │
│   388 │   │   │   │   self._buffer = None                                                        │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\windows_events.py:82 │
│ 6 in _poll                                                                                       │
│                                                                                                  │
│   823 │   │   │   # if the overlapped has been cancelled                                         │
│   824 │   │   │   elif not f.done():                                                             │
│   825 │   │   │   │   try:                                                                       │
│ ❱ 826 │   │   │   │   │   value = callback(transferred, key, ov)                                 │
│   827 │   │   │   │   except OSError as e:                                                       │
│   828 │   │   │   │   │   f.set_exception(e)                                                     │
│   829 │   │   │   │   │   self._results.append(f)                                                │
│                                                                                                  │
│ C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\windows_events.py:55 │
│ 2 in finish_send                                                                                 │
│                                                                                                  │
│   549 │   │                                                                                      │
│   550 │   │   def finish_send(trans, key, ov):                                                   │
│   551 │   │   │   try:                                                                           │
│ ❱ 552 │   │   │   │   return ov.getresult()                                                      │
│   553 │   │   │   except OSError as exc:                                                         │
│   554 │   │   │   │   if exc.winerror in (_overlapped.ERROR_NETNAME_DELETED,                     │
│   555 │   │   │   │   │   │   │   │   │   _overlapped.ERROR_OPERATION_ABORTED):                  │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
BrokenPipeError: [WinError 109] The pipe has been ended

Exception ignored in: <function BaseSubprocessTransport.__del__ at 0x0000022A33789240>
Traceback (most recent call last):
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\base_subprocess.py", line 125, in __del__
    _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\base_subprocess.py", line 80, in __repr__
    info.append(f'stderr={stderr.pipe}')
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\proactor_events.py", line 80, in __repr__
    info.append(f'fd={self._sock.fileno()}')
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\windows_utils.py", line 102, in fileno
    raise ValueError("I/O operation on closed pipe")
ValueError: I/O operation on closed pipe
Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x0000022A3378AD40>
Traceback (most recent call last):
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\proactor_events.py", line 116, in __del__
    _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\proactor_events.py", line 80, in __repr__
    info.append(f'fd={self._sock.fileno()}')
  File "C:\Users\joshua.janicas\AppData\Local\Programs\Python\Python310\lib\asyncio\windows_utils.py", line 102, in fileno
    raise ValueError("I/O operation on closed pipe")
ValueError: I/O operation on closed pipe
b
BatchConfig.__init__() missing 2 required positional arguments: 'encoding' and 'storage'
🧙‍♂️ 1
j
im guessing we can also include those into the batch_config ?
i suppose whatever values the default 10K provides...but i gotta find that
b
There is a default batch config the append
Copy code
DEFAULT_BATCH_CONFIG = {
    "encoding": {"format": "jsonl", "compression": "gzip"},
    "storage": {"root": "file://"},
}
Give this a try
Copy code
plugins:
  loaders:
  - name: target-snowflake
    variant: meltanolabs
    pip_url: meltanolabs-target-snowflake
    config:
      add_record_metadata: false # Can enable if we want more metadata
      #account: See .ENV
      #database: TS See .ENV 
      #user: See .ENV
      #role: See .ENV
      #warehouse: See .ENV
      #password: See .ENV
      default_target_schema: Raw # ${MELTANO_EXTRACT__LOAD_SCHEMA} # Meltano chooses the schema based on the `name` of the extractor
      hard_delete: false
      batch_size_rows: 150000
      batch_config:
        batch_size: 150000
        encoding:
          format: jsonl
          compression: gzip
        storage:
          root: "file://"
j
Copy code
batch_config:
        batch_size: 150000
        encoding: {"format": "jsonl", "compression": "gzip"}
        storage: {"root": "file://"}
b
I think that is a good sign?
j
yesss
b
ok cool. Then YAY!!!!
j
aw it didnt copy the gif, but its the high five one haha
thank so much
b
You are welcome. Glad I could help.
j
It went from taking 6 minutes to run 1 million rows to 1 minute
🎉 1
b
Very Cool. I think you can remove 
batch_size_rows
and it will still work. I hate to say that but I think it will.
j
a reasonable assumption, let me see
nope, we still need it. If I remove it it reverts back to 10 K chunks, because I bet you that's what it defaults to drawing from the source
b
Hmm that shocks me to be honest
@Edgar Ramírez (Arch.dev) we have been working with 
target_snowflake
and found we need to add in the 
batch_config
and 
batch_size_rows
to change how many rows get written to a batch file. Here is an example of the config below. I was wondering if this would be and SDK change or Target change to make this simpler for the user?
Copy code
plugins:
  loaders:
  - name: target-snowflake
    variant: meltanolabs
    pip_url: meltanolabs-target-snowflake
    config:
      add_record_metadata: false # Can enable if we want more metadata
      #account: See .ENV
      #database: TS See .ENV 
      #user: See .ENV
      #role: See .ENV
      #warehouse: See .ENV
      #password: See .ENV
      default_target_schema: Raw # ${MELTANO_EXTRACT__LOAD_SCHEMA} # Meltano chooses the schema based on the `name` of the extractor
      hard_delete: false
      batch_size_rows: 150000
      batch_config:
        batch_size: 150000
        encoding:
          format: jsonl
          compression: gzip
        storage:
          root: "file://"
I wondering if it is a easy as changing the 
DEFAULT_BATCH_CONFIG
for 
target-snowflake
to look like this and move it inside the 
SnowflakeSink
class?
Copy code
DEFAULT_BATCH_CONFIG = {
    "batch_size": self.max_size
    "encoding": {"format": "jsonl", "compression": "gzip"},
    "storage": {"root": "file://"},
}
e
Yeah, that probably makes sense. I think some confusion seeped in the target-snowflake implementation 😅, it shouldn't be using the batch config to decide how many rows to put in the temporary batch files it uses to load data. I don't think users need or want to control how target-snowflake loads its data beyond the batch size 🤷‍♂️.
b
Oh, I wonder if it takes input from the 
batch_config
in the 
meltano.yml
because it passes 
self.batch_config
when it creates 
batcher
. I know there is a property for 
batch_config
in the 
SnowflakeSink
class but I don't think the property is used if the 
batch_config
is present in the 
meltano.yml
?
Copy code
# serialize to batch files and upload
        # TODO: support other batchers
        batcher = JSONLinesBatcher(
            tap_name=self.target.name,
            stream_name=self.stream_name,
            batch_config=self.batch_config,
        )
e
Yeah, 
JSONLinesBatcher
should be getting a batch config object built on the fly with the value of 
batch_size_rows
and the other jsonl default values.
5 Views
Open in Slack
PreviousNext
Powered by