Как переносить данные откуда угодно куда угодно с помощью Meltano

                    pip3 install pipx
pipx ensurepath
pipx install cookiecutter
pipx install poetry
                

                    cookiecutter https://github.com/meltano/sdk --directory="cookiecutter/tap-template"
                

                    class YandexDirectStream(RESTStream):
   """YandexDirect stream class."""

   url_base = "https://api.direct.yandex.com/json/v5"

   records_jsonpath = "$.result[*]"  # Or override `parse_response`.
   next_page_token_jsonpath = "$.next_page"  # Or override `get_next_page_token`.
   def validate_response(self, response):
       if response.status_code == 400:
           data = response.json()
           raise FatalAPIError(f"Error message found: {data['error']['error_string']} {data['error']['error_detail']}")
       super().validate_response(response)

       if response.status_code in [201, 202]:
           raise RetriableAPIError("The report is being generated in offline mode")
       try:
           data = response.json()
           if data.get("error"):
                   raise FatalAPIError(f"Error message found: {data['error']['error_detail']}")
       except JSONDecodeError:
           if 200 <= response.status_code < 300:
               pass
  
   def backoff_max_tries(self) -> int:
       return 8
                

                    class CampaignsStream(YandexDirectStream):
   """Define custom stream."""

   name = "campaigns"
   path = "/campaigns"
   primary_keys = ["Id"]
   replication_key = None
   records_jsonpath = "$.result.Campaigns[*]"
   schema = th.PropertiesList(
       th.Property("Name", th.StringType),
       th.Property("Id", th.IntegerType, description="The user's system ID"),
   ).to_dict()

   def prepare_request_payload(
       self, context: Optional[dict], next_page_token: Optional[Any]
   ) -> Optional[dict]:
       data = {
           "method": "get",
           "params": {"SelectionCriteria": {}, "FieldNames": ["Id", "Name", "Status"]},
       }
       return data
                

                    from typing import List

from singer_sdk import Tap, Stream
from singer_sdk import typing as th  # JSON schema typing helpers
# TODO: Import your custom stream types here:
from tap_yandexdirect.streams import (
   YandexDirectStream,
   CampaignsStream,
)

STREAM_TYPES = [
   CampaignsStream,
]


class TapYandexDirect(Tap):
   """YandexDirect tap class."""
   name = "tap-yandexdirect"

   # TODO: Update this section with the actual config values you expect:
   config_jsonschema = th.PropertiesList(
       th.Property(
           "access_token",
           th.StringType,
           required=True,
           description="The token to authenticate against the API service"
       ),
       th.Property(
           "start_date",
           th.DateTimeType,
           description="The earliest record date to sync"
       ),
       th.Property(
           "end_date",
           th.DateTimeType,
           description="The earliest record date to sync"
       ),
   ).to_dict()

   def discover_streams(self) -> List[Stream]:
       """Return a list of discovered streams."""
       return [stream_class(tap=self) for stream_class in STREAM_TYPES]
                

                    poetry run tap-yandexdirect
                

                    meltano init my-project
                

                    meltano add --custom extractor tap-yandexdirect 
# (namespace) [tap_yandexdirect]: <hit enter, no value needed>
# (pip_url) [tap-yandexdirect]: git+https://github.com/epoch8/tap-yandexdirect.git 
# (executable) [pipelinewise-tap-s3-csv]: tap-yandexdirect
# (capabilities) [[]]: properties,discover,state
Затем добавьте target-bigquery:
meltano add loader target-bigquery
                

                    meltano --log-level=debug invoke tap-yandexdirect --help 
meltano --log-level=debug invoke target-bigquery --help
                

                    meltano config tap-yandexdirect set start_date 2021-11-02 

meltano config  tap-yandexdirect set end_date 2022-01-01

meltano config  tap-yandexdirect set access_token your-yandexdirect-token
                

                    meltano config target-bigquery set project_id {your_project_id}

meltano config target-bigquery set dataset_id {your_dataset_id}

meltano config target-bigquery set location {your_dataset_location}

meltano config target-bigquery set credentials_path {your_path_to_service_account_credentials.json}
                

                    version: 1
default_environment: dev
project_id: e564ac62-b9ed-41c5-8941-083c535b50eb
plugins:
 extractors:
 - name: tap-yandexdirect
   namespace: tap_yandexdirect
   pip_url: git+https://github.com/epoch8/tap-yandexdirect.git
   executable: tap-yandexdirect
   capabilities:
   - state
   - discover
   - catalog
   config:
     access_token: your-access-token
     start_date: 2021-11-02
     end_date: 2022-01-01
 loaders:
 - name: target-bigquery
   variant: adswerve
   pip_url: git+https://github.com/adswerve/target-bigquery.git
   config:
     project_id: your_project_id
     dataset_id: your_dataset_id
     location: your_dataset_locations
     credentials_path: your_path_to_service_account_credentials

environments:
- name: dev
- name: staging
- name: prod