[feat]deploy with k8s

.gitignore

@@ -117,4 +117,6 @@ pip-selfcheck.json
 .envrc
 .direnv
 
-.pylintrc
+.pylintrc
+
+airflow-ssh-secret.yaml

Dockerfile

@@ -8,11 +8,11 @@ RUN apt-get update \
   && apt-get clean \
   && rm -rf /var/lib/apt/lists/*
 
-RUN mkdir -p /lakehouse/airlake
-COPY --chown=airflow:airflow . /lakehouse
+RUN mkdir -p /bigdata/airlake
+COPY --chown=airflow:airflow . /bigdata
 COPY requirements.txt /
 
-ENV PYTHONPATH=$PYTHONPATH:/lakehouse
+ENV PYTHONPATH=$PYTHONPATH:/bigdata
 
 USER airflow
 RUN pip install --no-cache-dir -r /requirements.txt

airlake/operators/bq_to_elasticsearch.py

@@ -1,16 +1,13 @@
 from airflow.models import BaseOperator
-from airflow.hooks.base import BaseHook
 from airflow.utils.decorators import apply_defaults
 from airflow.providers.google.cloud.hooks.gcs import GCSHook as GoogleCloudStorageHook
 from airflow.providers.google.cloud.hooks.bigquery import BigQueryHook
 
-
-from elasticsearch import Elasticsearch
-from elasticsearch import helpers
+from airflow.utils.context import Context
+from typing import Any
 import gzip
 import json
 import os
-import io
 from airlake.hooks.elasticsearch_hook import ElasticSearchHook
 
 
@@ -32,7 +29,7 @@ def __init__(
         sql="",
         destination_cloud_storage_uris="",
         source_project_dataset_table="",
-        source_bucket="tiki-brain",
+        source_bucket="gcs-stg",
         es_index="",
         gcp_conn_id="bigquery_default",
         gcs_bucket="",
@@ -75,7 +72,7 @@ def __init__(
         self.size_bulk = size_bulk
         # validate input from config dag
 
-    def execute(self, context):
+    def execute(self, context: Context) -> Any:
         self.log.info(
             "Sync table %s to elasticsearch", self.source_project_dataset_table
         )

airlake/operators/bq_to_gcs.py

@@ -1,2 +1,28 @@
-class BigqueryToGcsOperator:
-    pass
+from typing import Any
+from airflow.models import BaseOperator
+from airflow.utils.context import Context
+
+class BigqueryToGcsOperator(BaseOperator):
+    template_fields = (
+        "source_project_dataset_table",
+        "destination_cloud_storage_uris",
+        "gcs_bucket",
+        "gcs_prefix",
+        "gcp_conn_id",
+        "labels",
+    )
+    def __init__(
+        self,
+        source_project_dataset_table: str = None,
+        destination_cloud_storage_uris: str = None,
+        compression: str = "",
+        export_format: str = "CSV",
+    ):
+        super(BigqueryToGcsOperator, self).__init__()
+        self.source_project_dataset_table = source_project_dataset_table
+        self.destination_cloud_storage_uris = destination_cloud_storage_uris
+        self.compression = compression
+        self.export_format = export_format
+
+    def execute(self, context: Context) -> Any:
+        return super().execute(context)

airlake/operators/bq_to_sheet.py

@@ -1,2 +1,27 @@
-class BigqueryToSheetOperator:
-    pass
+from typing import Any
+from airflow.models import BaseOperator
+from airflow.utils.context import Context
+
+class BigqueryToSheetOperator(BaseOperator):
+    template_fields = (
+        "source_project_dataset_table",
+        "destination_cloud_storage_uris",
+        "gcs_bucket",
+        "gcs_prefix",
+        "gcp_conn_id",
+        "labels",
+    )
+    def __init__(
+        self,
+        source_project_dataset_table: str = None,
+        destination_cloud_storage_uris: str = None,
+        compression: str = "",
+        export_format: str = "CSV",
+    ):
+        self.source_project_dataset_table = source_project_dataset_table
+        self.destination_cloud_storage_uris = destination_cloud_storage_uris
+        self.compression = compression
+        self.export_format = export_format
+
+    def execute(self, context: Context) -> Any:
+        return super().execute(context)