I migrated a Fivetran pipeline to dltHub for $0.65 A developer replaced a daily Fivetran MySQL sync with a dltHub pipeline in about an hour, spending $0.65 in AI tokens using OpenCode with Minimax M3. The migration required five manual inputs, including adding an SSH key and verifying the plan, and the new pipeline triggers a Snowflake Cortex Search refresh after load. Lakehouse https://www.sfrt.io/tag/lakehouse/ I migrated a Fivetran pipeline to dltHub for $0.65 One hour, five manual inputs, and $0.65 of tokens... enough to replace a daily Fivetran sync with a tested dltHub job TL/DR: I replaced one daily Fivetran MySQL sync with dltHub in about an hour. OpenCode running Minimax M3 because multimodal and still cheap 😅 consumed $0.65 in tokens. I gave it five manual inputs. One was actual infrastructure work. The other four were decisions and verification. The old connection pulled 12 CMS backend our Drupal uses MySQL tables from a webserver through an SSH tunnel into Snowflake. It worked. And it was billed via monthly-active-rows 🙄 And it felt like a reasonable candidate for a small experiment: could I recreate the operational behavior with dltHub, use OpenCode as the implementation environment, and stay in a range where an hour of AI usage costs less than a coffee in Zurich? Turns out... yes. With caveats, obviously. The baseline I first had to log in to Fivetran for the first time in months it is a reliable platform after all to get the connector ID. Then I used the Fivetran MCP https://github.com/fivetran/fivetran-mcp?ref=sfrt.io to inspect its MySQL configuration and how it connects to the webserver through an SSH tunnel. Its selected 12 tables are all part of a Drupal CMS, and it runs daily. I did not want a generic MySQL ingestion example. I wanted the existing job's behavior: same source scope, data landing in Snowflake, incremental loads where I could justify them, and a downstream task triggered ❄️ Cortex Search refresh after a successful load. Five manual inputs I did not sit back and accept generated code blindly. The migration needed five interventions. - Add a new SSH key: This was the major manual task. I created a dedicated key pair and added its public key to the servive user's authorized keys file on the webserver. I also captured the server's host key while I was there so the new job can pin it instead of trusting whatever answers on port 22. - Decide source scope: The first profile discovered 346 tables and roughly 6.2 million rows. That was technically fun and operationally wrong. The previous Fivetran configuration had 12 enabled tables, because those are what I actually use. So I told OpenCode to match that scope exactly. - Verify the plan: I checked the generated design against the source schema and test suite. In particular, I wanted primary keys reflected from MySQL, a conservative cursor whitelist for incremental loads: I didn't define the PK and cursor for each of the 12 tables individually but rather used a dltHub skill https://github.com/dlt-hub/dlthub-ai-workbench?ref=sfrt.io to annotate sources and identify usually useful cursors , and cleanup of the SSH tunnel. - Add a downstream task: Fivetran stopped after the load. I asked OpenCode to trigger the existing Snowflake Cortex Search refresh task after dlt completes. The new pipeline now does a little more work than its predecessor 😎 - Verify the run: After everything was implemented and tested locally, I had to point OpenCode to a modification of the pipeline necessary for it to run on dltHub cf. below . What dltHub generated dlt's sql database source doesn't open SSH tunnels itself https://dlthub.com/docs/dlt-ecosystem/verified-sources/sql database/configuration?ref=sfrt.io connecting-to-a-remote-database-over-ssh . Hence, my new job opens a pinned tunnel, then creates a SQLAlchemy engine through its ephemeral local port, and passes that engine to dlt. sshtunnel convenience wrapper around Paramiko recommended in the dltHub docs is not maintained since 2021 and not compatible with recent Paramiko versions. Hence, I use a custom Paramiko-based local TCP forwarder ParamikoTunnel instead of the SSHTunnelForwarder mentioned in the docs 😜 python def get ssh tunnel ssh credentials : private key = paramiko.Ed25519Key.from private key io.StringIO ssh credentials "private key" return ParamikoTunnel hostname=ssh credentials "server ip address" , username=ssh credentials "username" , private key=private key, host key=parse ssh host key ssh credentials "host key" , target address= "127.0.0.1", 3306 , The database has composite primary keys everywhere. I reflected MySQL metadata and only configured incremental merge for tables with a primary key plus one of three mutation-oriented columns: SAFE CURSOR PRIORITY = "changed", "content translation changed", "revision timestamp", def apply resource hints table name, resource, table : pk columns = column.name for column in table.primary key.columns cursor = choose cursor table name, column.name for column in table.columns if cursor is None or not pk columns: resource.apply hints write disposition="replace" return resource.apply hints write disposition="merge", primary key=pk columns, incremental=dlt.sources.incremental cursor, on cursor value missing="include" , I deliberately rejected created , generic timestamp , and revision IDs as universal cursors. Drupal puts lots of time-related integers into its schema. A plausible name is not enough evidence for incremental correctness. Two failures worth mentioning The first remote run failed and there goes my 100% pipeline-run-success-rate for the week 😤 because dltHub connected to Snowflake from an IP outside the account allowlist. I wrapped the job with my existing Snowflake proxy context used by other dltHub jobs. On the source-side of the pipeline, SSH and MySQL keep using raw TCP, so they are unaffected by those HTTP proxy variables. static egress IPs https://dlthub.com/docs/devel/hub/pipeline-operations/job-configuration?ref=sfrt.io static-egress-ips now 😎 I continue using my proxy as it also is a gateway of some other connections anyway. The first all-table run also exhausted SQLAlchemy's connection pool. I had given dlt hundreds of resources and long-lived MySQL reads. A larger pool only delayed the failure. NullPool fixed it by opening and closing a connection per source read: return sa.create engine url, poolclass=sa.pool.NullPool One dependency issue also appeared here: The aforementioned sshtunnel calls Paramiko's removed DSSKey API, so I build a custom ParamikoTunnel instead. The smoke test found that before deployment, which is exactly why I keep smoke tests around. Proof, then schedule The scoped baseline run completed locally in 31 seconds. The second run restored state from Snowflake and completed in 16 seconds, confirming that dlt retained its cursors. The deployed dltHub job completed in 39 seconds with zero failed jobs. And I really don't care about the number of processed rows anymore, as the only measure relevant for billing now is runtime 😜 I also kept the verification close to the implementation: 25 focused tests cover SSH host-key parsing, disabled SSH-agent lookup, exact table scope, composite keys, cursor selection, cleanup, incremental state, and downstream SQL. What I got for $0.65 I got a working dltHub job, focused tests, deployment configuration, a documented secret shape, and a better downstream behavior than the old Fivetran sync. I still had to make the data and security decisions. That part should remain human work. But I spent an hour turning five pieces of manual context into a deployed pipeline instead of typing every tunnel option, test fixture, and metadata rule myself. That is a pretty good trade for $0.65 😎 The full pipeline: python """Load 12 enabled Fivetran tables into Snowflake over SSH.""" import base64 import io import logging import select import socket import sys import threading import time import dlt import paramiko from dlt.sources.sql database import sql database import sqlalchemy as sa logger = logging.getLogger name SAFE CURSOR PRIORITY = "changed", "content translation changed", "revision timestamp", TABLE NAMES = "media field image caption", "media revision field image caption", "node body", "node field shared paragraphs", "node field data", "paragraph field pg imgtxt text", "paragraph field pg media text", "paragraph field pg testimonial testimonial", "paragraph field pg text text", "path alias", "taxonomy index", "taxonomy term field data", def parse ssh host key host key : """Parse an OpenSSH ed25519 host-key line in memory.""" key type, encoded key, = host key.split if key type = "ssh-ed25519": raise ValueError f"unsupported SSH host-key type: {key type}" return paramiko.Ed25519Key data=base64.b64decode encoded key, validate=True class ParamikoTunnel: """Forward local TCP clients to a remote TCP service over SSH.""" MAX WORKERS = 64 def init self, hostname, username, private key, host key, target address : self.hostname = hostname self.username = username self.private key = private key self.target address = target address self. client = paramiko.SSHClient self. client.get host keys .add hostname, host key.get name , host key self. client.set missing host key policy paramiko.RejectPolicy self. listener = None self. accept thread = None self. workers = set self. connections = set self. lock = threading.Lock self. lifecycle lock = threading.RLock self. stop event = threading.Event self. stopping = False self. worker slots = threading.BoundedSemaphore self.MAX WORKERS self.local bind port = None def start self : """Connect SSH and start a local listener on an ephemeral port.""" with self. lifecycle lock: if self. listener is not None: return try: self. client.connect hostname=self.hostname, port=22, username=self.username, pkey=self.private key, allow agent=False, look for keys=False, except Exception: self. client.close raise listener = socket.socket socket.AF INET, socket.SOCK STREAM try: listener.setsockopt socket.SOL SOCKET, socket.SO REUSEADDR, 1 listener.bind "127.0.0.1", 0 listener.listen listener.settimeout 0.2 except Exception: listener.close self. client.close raise with self. lock: self. stop event.clear self. stopping = False self. listener = listener self.local bind port = listener.getsockname 1 self. accept thread = threading.Thread target=self. accept loop, name="paramiko-tunnel-accept", daemon=True, self. accept thread.start def accept loop self : listener = self. listener while not self. stop event.is set : try: client socket, client addr = listener.accept except socket.timeout: continue except OSError: break with self. lock: stopping = getattr self, " stopping", False if stopping or self. stop event.is set or not self. worker slots.acquire blocking=False : try: client socket.close except Exception: pass continue worker = threading.Thread target=self. forward connection, args= client socket, client addr , name="paramiko-tunnel-forward", daemon=True, with self. lock: if getattr self, " stopping", False or self. stop event.is set : self. worker slots.release reject = True else: reject = False self. connections.add client socket self. workers.add worker try: worker.start except Exception: self. connections.discard client socket self. workers.discard worker self. worker slots.release try: client socket.close except Exception: pass logger.exception "Paramiko tunnel worker failed to start" if reject: try: client socket.close except Exception: pass continue def forward connection self, client socket, client addr : channel = None try: try: channel = self. client.get transport .open channel "direct-tcpip", self.target address, client addr with self. lock: self. connections.add channel self. relay client socket, channel, self. stop event except Exception: if not self. stop event.is set : logger.exception "Paramiko tunnel worker failed" finally: for connection in client socket, channel : if connection is not None: try: connection.close except Exception: pass with self. lock: self. connections.discard connection with self. lock: self. workers.discard threading.current thread self. worker slots.release @staticmethod def relay client socket, channel, stop event=None : sources = {client socket, channel} while sources: try: readable, , = select.select list sources , , , 0.2 except Exception: if stop event is None or not stop event.is set : logger.exception "Paramiko tunnel relay select failed" return if not readable: continue for source in readable: try: payload = source.recv 65536 except Exception: if stop event is None or not stop event.is set : logger.exception "Paramiko tunnel relay receive failed" return if not payload: destination = channel if source is client socket else client socket try: if destination is channel: destination.shutdown write else: destination.shutdown socket.SHUT WR except Exception: pass sources.remove source continue destination = channel if source is client socket else client socket try: destination.sendall payload except Exception: if stop event is None or not stop event.is set : logger.exception "Paramiko tunnel relay send failed" return def stop self : """Stop accepting clients and close SSH, listener, and forwarded sockets.""" with self. lifecycle lock: with self. lock: self. stopping = True self. stop event.set listener = self. listener self. listener = None if listener is not None: try: listener.close except Exception: pass with self. lock: connections = list self. connections for connection in connections: try: connection.close except Exception: pass try: self. client.close finally: deadline = time.monotonic + 5 while True: with self. lock: connections = list self. connections workers = list self. workers for connection in connections: try: connection.close except Exception: pass threads = self. accept thread, workers for thread in threads: if thread is not None and thread is not threading.current thread : thread.join timeout=0.1 if not any thread is not None and thread.is alive for thread in threads : break if time.monotonic = deadline: break self. accept thread = None self.local bind port = None def get ssh tunnel ssh credentials : """Build an SSH tunnel without opening the connection.""" private key = paramiko.Ed25519Key.from private key io.StringIO ssh credentials "private key" return ParamikoTunnel hostname=ssh credentials "server ip address" , username=ssh credentials "username" , private key=private key, host key=parse ssh host key ssh credentials "host key" , target address= "127.0.0.1", 3306 , def create mysql engine db credentials, local port : """Create a MySQL engine connected to the tunnel's local endpoint.""" url = sa.URL.create drivername=db credentials.get "drivername", "mysql+pymysql" , username=db credentials "username" , password=db credentials "password" , host="127.0.0.1", port=local port, database=db credentials.get "database", "