Repository navigation
Enable Streaming from SFTP to GCS in SFTPToGCSOperator #34995
Description
Description
The current SFTPToGCSOperator in Airflow does not support streaming objects directly from SFTP to Google Cloud Storage (GCS). This design poses a challenge since both Composer and Airflow best practices recommend against assigning large amounts of memory to workers. Transferring sizeable files from SFTP to GCS without streaming can potentially max out worker memory.
Use case/motivation
Prevent worker memory exhaustion, especially when dealing with large files.
Related issues
No response
Are you willing to submit a PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct
Activity
boring-cyborg commentedon Oct 17, 2023
Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.
[-]Enhancement: Enable Streaming from SFTP to GCS in `SFTPToGCSOperator`[/-][+]Enable Streaming from SFTP to GCS in `SFTPToGCSOperator`[/+]hegongwork commentedon Oct 17, 2023
hi @ferruzzi
yesterday in my project I wrote this code and my colleague @rynmccrmck told me I could added the code to Airflow and this PR is his draft for me. I will create a same PR.
ferruzzi commentedon Oct 17, 2023
I'll assign the issue to you then. let me know if you need any help
hegongwork commentedon Oct 18, 2023
update: I wrote a PR in my own repo: https://github.com/he-gong-sardine/airflow/pull/1/files
but I haven't set up airflow environment so I cannot test it.
I will send PR after I test it locally.
Player00liv commentedon Oct 23, 2023
Vhuut
elaye-canopy commentedon Dec 21, 2023
@he-gong-sardine did you get a chance to test it?
I am encountering a similar issue and was in the process of testing a fix independent of yours (I will also upload it later for comparison), but if yours already works I might just go with yours.
I've also left a comment (well, a question, but not a criticism) on your PR.
Also, not directly related but logically related - perhaps similar logic should be made for the converse operator as well (GCS to SFTP)
hegongwork commentedon Jan 3, 2024
@elaye-canopy
elaye-canopy commentedon Jan 7, 2024
@he-gong-sardine
Great to see this is worked on, we independently solved it on our side and I put a comment in the code about that.
Your solution does seem to work but seems to have led to significant slowdown when we tested it on files > 60gb (we obviously didn't test the google upload from file version on that file size).
I put a comment with a snippet of the code we used to solve it, as well as a link to Paramiko's documentation where you can find the signature of the Paramiko maintained methods (getfo) which uses Paramiko's prefetch to solve the performance issues.
We have a DAG downloading 4 files simultaneously (10GB, 16GB, 60GB, and 166GB) so a slowdown would have meant we would keep doing it all day long.
The file is in a private repo currently but if you want my version of the fix, to make it PR ready, feel free to contact me and I'll get it to you.
elaye-canopy commentedon Jan 7, 2024
Also, again, not directly related but logically related - perhaps similar logic should be made for the converse operator as well (GCS to SFTP).
There is a paramiko funciton putfo similiar to the getfo that I recommended in my comment, which could be useful for that.
Vamsi-klu commentedon Jun 13, 2026
Vamsi-klu commentedon Aug 27, 2026
Honor gzip and mime_type on SFTPToGCS stream path (#72132)
Honor gzip and mime_type on SFTPToGCS stream path (apache#72132)