Depuis que j'ai implémenté Airflow dans mon étude personnelle, j'écrirai ce que j'ai remarqué à ce moment-là [^ 1]. J'espère que moins de gens sont dépendants de problèmes similaires.
--Utiliser l'exécuteur local
AIRFLOW_EXTRAS Il s'agit d'un plug-in pour étendre le flux d'air, et il existe un accès à GCP à partir d'un système de base de données tel que MySQL. Le Dockerfile officiel dit:
ARG AIRFLOW_EXTRAS="async,aws,azure,celery,dask,elasticsearch,gcp,kubernetes,mysql,postgres,redis,slack,ssh,statsd,virtualenv"
crypto est pratiquement nécessaire car il est nécessaire pour générer FERNET_KEY.
Puisque nous utilisons MySQL pour la base de données principale et psycopg2 pour nous connecter à Redshift, nous avons également besoin de quelque chose lié à ceux-ci.
entrypoint.sh
Comme vous pouvez le voir dans la documentation (https://airflow.apache.org/docs/stable/howto/secure-connections.html) et puckel, vous pouvez maintenant générer une FERNET_KEY pour crypter votre connexion. À. Plus sûr que l'écriture solide dans ʻairflow.cfg`.
: "${AIRFLOW__CORE__FERNET_KEY:=${FERNET_KEY:=$(python -c "from cryptography.fernet import Fernet; FERNET_KEY = Fernet.generate_key().decode(); print(FERNET_KEY)")}}"
Même si vous faites depend_on: mysql dans docker-compose.yml, il attend juste que le conteneur démarre et ne confirme pas que la base de données démarre. puckel utilise la commande nc dans ʻentrypoint.sh` pour vérifier si une connexion à la base de données est établie. Ce genre de chose est très utile.
wait_for_port() {
local name="$1" host="$2" port="$3"
local j=0
while ! nc -z "$host" "$port" >/dev/null 2>&1 < /dev/null; do
j=$((j+1))
if [ $j -ge $TRY_LOOP ]; then
echo >&2 "$(date) - $host:$port still not reachable, giving up"
exit 1
fi
echo "$(date) - waiting for $name... $j/$TRY_LOOP"
sleep 5
done
}
Définissez les variables d'environnement suivantes (Référence). Ce sont les paramètres du côté DB, mais ils sont accessibles depuis Airflow en utilisant le même mot de passe utilisateur.
my.cnf
Définition de ʻexplicit_defaults_for_timestamp = 1` lors de l'utilisation de MySQL, comme décrit dans la description du backend de la base de données Airflow (https://airflow.apache.org/docs/stable/howto/initialize-database.html) Est nécessaire.
De plus, ajoutez des paramètres pour la gestion des caractères multi-octets.
[mysqld]
character-set-server=utf8mb4
explicit_defaults_for_timestamp=1
[client]
default-character-set=utf8mb4
AIRFLOW__CORE__SQL_ALCHEMY_CONN
La valeur par défaut est sqlite, mais changez-la en fonction de la base de données et du pilote. Pour savoir comment écrire, reportez-vous au [Document] de SqlAlchemy (https://docs.sqlalchemy.org/en/13/core/engines.html)
mysql+mysqldb://user:password@host:port/dbpostgresql+psycopg2://user:password@host:port/dbL'hôte est celui spécifié par container_name dans docker-compose.yml, et le port est essentiellement 3306 pour MySQL et 5432 pour PostgreSQL. Le nom d'utilisateur et le nom de la base de données sont ceux définis ci-dessus.
Comme décrit dans Documentation, les paramètres peuvent être écrits à plusieurs endroits et les variables d'environnement sont hiérarchisées. À. Je souhaite utiliser des variables d'environnement pour l'authentification AWS. Si vous voulez le rendre statique, comme accéder à la base de données, vous pouvez utiliser ʻairflow.cfg`. Au niveau de la production, les membres doivent définir correctement les règles.
Les plus élevés ont la priorité.
C'est également assez varié. Cela devrait être priorisé comme celui ci-dessous.
Dockerfile: utilisez celui avec peu de changements, comme celui par défautdocker-compose.yml: peut être modifié pour d'autres conteneurs. Plus flexible..env: S'il est spécifié comme ʻenv_file dans docker-compose.yml`, il sera lu au démarrage du conteneur. Écrivez ici le système d'authentification que vous ne souhaitez pas laisser dans Git.docker-compose.ymlÉtant donné que les mêmes paramètres sont utilisés, il est plus facile de gérer et de gérer si l'emplacement de la description est le même.
version: "3.7"
services:
mysql:
image: mysql:5.7
container_name: mysql
environment:
- MYSQL_ROOT_PASSWORD=password
- MYSQL_USER=airflow
- MYSQL_PASSWORD=airflow
- MYSQL_DATABASE=airflow
volumes:
- ./mysql.cnf:/etc/mysql/conf.d/mysql.cnf:ro
ports:
- "3306:3306"
airflow:
build: .
container_name: airflow
depends_on:
- mysql
environment:
- AIRFLOW_HOME=/opt/airflow
- AIRFLOW__CORE__LOAD_EXAMPLES=False
- AIRFLOW__CORE__EXECUTOR=LocalExecutor
- AIRFLOW__CORE__SQL_ALCHEMY_CONN=mysql+mysqldb://airflow:airflow@mysql:3306/airflow
- MYSQL_PORT=3306
- MYSQL_HOST=mysql
#Abréviation
.envHormis Redshift, les clés d'accès et les clés secrètes AWS sont très sensibles, donc je ne veux pas les écrire dans docker-compose.yml ou ʻentrypoint.sh. ʻAirflow.cfg peut être envisagé, mais en réalité, il peut s'agir d'une consultation avec l'équipe de développement.
** Pour le moment, il n'est pas moderne de taper dans l'interface graphique. ** **
Lors de l'écriture, reportez-vous à la Documentation et écrivez comme suit.
| Conn Id | Conn Type | Login | Password | Host | Port | Schema | Environment Variable |
|---|---|---|---|---|---|---|---|
| redshift_conn_id | postgres | awsuser | password | your-cluster-host | 5439 | dev | AIRFLOW_CONN_REDSHIFT_CONN_ID=postgres://awsuser:password@your-cluster-host:5439/dev |
| aws_conn_id | aws | your-access-key | your-secret-key | AIRFLOW_CONN_AWS_CONN_ID=aws://your-access-key:your-secret-key@ |
Même si l'ID est faible, il sera élevé dans le nom de la variable d'environnement.
Pour les clés AWS, vous devez ajouter @ à la fin, même s'il n'y a pas d'hôte. Sinon, une erreur se produira. De plus, si la clé contient un deux-points ou une barre oblique, elle ne sera pas bien analysée, il est donc préférable de régénérer la clé.
Au contraire, je voulais connaître le format URI de la connexion entré dans l'interface graphique, etc. Vous pouvez sortir comme suit.
from airflow.hooks.base_hook import BaseHook
conn = BaseHook.get_connection('postgres_conn_id')
print(f"AIRFLOW_CONN_{conn.conn_id.upper()}='{conn.get_uri()}'")
Comme pour les connexions, vous pouvez Définir la valeur de clé. Comme méthode,
Pour le code
from airflow.models import Variable
Variable.set(key="foo", value="bar")
Pour les variables d'environnement
| Key | Value | Environment Variable |
|---|---|---|
| foo | bar | AIRFLOW_VAR_FOO=bar |
Même si la clé est inférieure, le nom de la variable d'environnement sera supérieur.
Pour le moment, je voudrais vous présenter le Repository.
[^ 1]: Je suis un ingénieur de données d'Udacity. J'ai touché Cassandra, Redshift, Spark, Airflow. On disait que cela prendrait 5 mois, mais ça s'est terminé dans 3 mois, il semble donc préférable de signer un contrat mensuel. De plus, vous bénéficierez de 50% de réduction sur une base régulière, il est donc recommandé de vous inscrire dans ce but. ~~ Sinon Takasugi ~~
[^ 2]: Quand j'ai touché ʻapache / airflow: 11.1010en écrivant l'article, cela semblait assez fluide. Si vous exécutezdocker run -it --name test -p 8080 -d apache / airflow: 11.1010" ", il commencera par bash open, donc il est flexible comme docker exec test airflow initdb`. Vous pouvez le faire fonctionner.
Recommended Posts