fpopic

# Copied from https://cloud.google.com/composer/docs/concepts/versioning/composer-versions for composer-2.0.2-airflow-2.1.4-python-3.8.12.
absl-py==1.0.0
alembic==1.7.3
amqp==2.6.1
anyio==3.3.1
apache-airflow==2.1.4+composer
apache-airflow-providers-apache-beam==3.1.0
apache-airflow-providers-cncf-kubernetes==3.0.0
apache-airflow-providers-ftp==2.0.1
apache-airflow-providers-google==6.3.0

fpopic

Use custom Python virtual environment

In order to not use System Python installation, we can use custom virtual environment tools (e.g. virtualenv, pyenv, ...) that copy and isolate System Python installation into project folder so that installation of custom packages are done per project/repository basis.

• Install virtualenv (brew install virtualenv)

  cd some-python-gitlab-repo
  

create virtual environment

fpopic

WITH

t AS (
  SELECT 
    '20190101' AS _TABLE_SUFFIX,
    DATE '2019-01-01' AS date,
    1 AS fullVisitorId,
    1 AS visitId,
    [
      STRUCT(

fpopic

WITH 

T_A AS (
  SELECT STRUCT("1"  AS x,  2 AS y) a -- schema a is huge
),

T_B AS(
  SELECT STRUCT(TRUE  AS z,  4.0 AS w) b -- schema b is huge as well
)

fpopic

From @echojc's awesome blog post and followup

fpopic

// Would like to serialize any Scala case class (could be nested as well) to Map[String, Any] 

case class N(j: Int)

case class A(i: Int, s: Some[Int], l: List[Int], n: N)

// Want to generate:

object N extends Mappable[N] {
  def toMap(n: N): Map[String, Any] = {

fpopic

1. Producer API

- partitionsFor(topic)

// transactional producer

- initTransaction()  // uses config transactional.id
- beginTransaction()
- commitTransaction()

fpopic

scala> Seq(1,2,3).map{x=>println(x); x*100}.map{x=>println(x); x/10}
1
2
3
100
200
300
res5: Seq[Int] = List(10, 20, 30)

scala> Seq(1,2,3).view.map{x=>println(x); x*100}.map{x=>println(x); x/10}.force

fpopic

import logging

from airflow.hooks.postgres_hook import PostgresHook
from airflow.models import BaseOperator
from airflow.utils.decorators import apply_defaults


class SelectPostgresOperator(BaseOperator):
    """
     Executes sql code in a specific Postgres database

fpopic

# SETUP AEROSPIKE NAMESPACE (to use hdd)
sudo gedit /etc/aerospike/aerospike.conf

```
namespace test-namespace {
	replication-factor 2		# Each element has one copy
	memory-size 1G			# Maximum memory allocation for primary and secondary indexes.
	default-ttl 30d			# Bin expiration 	 

	# storage-engine memory 	# a) Configure the storage-engine to use in-memory only