[PYTHON] Verwenden Sie Cloud Composer, um regelmäßig auf die Youtube-API zuzugreifen und eine Pipeline zum Speichern der Ergebnisse in Bigquery zu erstellen
Einführung
Hallo. Ich bin Datenwissenschaftler bei einem KI-Unternehmen. Vor kurzem bin ich süchtig nach Youtubers Tokai On Air, aber plötzlich habe ich mich gefragt, wie schnell sie die Gesamtzahl der Aufrufe und die Anzahl der Registranten erhöhen. Deshalb habe ich die Youtube-API regelmäßig mit Cloud Composer aufgerufen und sie regelmäßig aufgerufen. Ich habe einen Code geschrieben, um tägliche numerische Übergänge in Bigquery zu speichern.
Übrigens können Sie den Übergang der Anzahl der Ansichten für den letzten Monat von den folgenden Websites aus sehen, aber ich wollte die Daten für einen längeren Zeitraum sammeln und analysieren, also mache ich diesmal. [Einfach] 3 Möglichkeiten, um den Übergang der Anzahl der Abonnenten des YouTube-Kanals zu überprüfen
Grundeinstellungen der Youtube-API
Nehmen Sie die ersten Einstellungen vor, damit Sie auf die Youtube-API zugreifen können. Grundsätzlich können Sie damit umgehen, indem Sie im Element "Dienste" von GCP nach der YouTube-Daten-API suchen und diese aktivieren. Speichern Sie danach den Youtube-API-Schlüssel. Informationen zu diesem Bereich finden Sie auf der offiziellen Website.
Einstellungen der GCP-Dienstumgebung
Es ist kein Problem, wenn Sie verschiedene Einstellungen für die folgende Site vornehmen. Ich verwende jedoch die Benutzeroberfläche, um Cloud Composer und Bigquery zu konfigurieren, da die Konfiguration über die Benutzeroberfläche einfacher ist. Verfahren zum Erstellen eines Workflows zum Bearbeiten von BigQuery-Tabellen in GCP Cloud Composer
Cloud Composer
Stellen Sie nur die folgenden Teile ein und drücken Sie "Erstellen", ohne etwas anderes einzustellen. Das Erstellen eines Komponisten nimmt viel Zeit in Anspruch.
Bibliotheksinstallation
Fahren Sie nach dem Erstellen des Komponisten mit den folgenden Aufgaben fort. Wenn die Youtube-API nicht auf dem Composer installiert ist, können Sie die Youtube-API später nicht mehr vom Composer aus aufrufen. Installieren Sie daher die Youtube-API im Voraus auf dem Composer. Erstellen Sie zunächst die folgende Inhaltsdatei.
requirement.txt
youtube-data-api
Laden Sie diese Datei in die Cloud Shell hoch. Sie können hochladen, indem Sie auf die [offizielle Website] verweisen (https://cloud.google.com/blog/ja/products/gcp/google-cloud-shell-10-tips). Führen Sie dann den folgenden Code in der Cloud Shell aus.
gcloud config set core/Projekt Projektname
gcloud composer-Umgebungen aktualisieren den Komponisten-Namen(toukai-onair-composer) --update-pypi-packages-from-file requirements.txt --Standort Standortname(asia-northeast1)
Bigquery Erstellen Sie die Tabelle wie auf der folgenden Site beschrieben. Erstellen und Verwenden von Tabellen
Erstellen Sie eine DAG-Definition für Composer
Nachdem die Einstellungen so gut wie vorgenommen wurden, klicken Sie auf die Youtube-API und schreiben Sie den Code, der in Bigquery gespeichert werden soll.
Code, der auf die Youtube-API trifft
Unten finden Sie den Code, der die Gesamtzahl der Aufrufe und Abonnenten für einen bestimmten Kanal unter Berücksichtigung von channel_id und api_key zurückgibt.
youtubelib.py
from apiclient.discovery import build
class Youtubeapi:
def __init__(self, channel_id, api_key):
self.channel_id = channel_id
if not channel_id:
raise Exception("need channel_id")
self.api_key = api_key
if not api_key:
raise Exception("need api_key")
def get_statistics_data(self):
youtube = build('youtube', 'v3', developerKey=self.api_key)
search_response = youtube.channels().list(
part='statistics',
id=self.channel_id,
).execute()
return search_response['items'][0]['statistics']
Übrigens wird channel_id im URL-Link aufgeführt, wenn Sie Ihren Lieblings-Youtuber auf Youtube durchgehen und dort eingeben. Wenn es sich um Tokai On Air handelt, lautet der Link wie folgt, sodass die channel_id "UCutJqz56653xV2wwSvut_hQ" lautet.
https://www.youtube.com/channel/UCutJqz56653xV2wwSvut_hQ
Sie können den api_key auch dann abrufen, wenn Sie die Youtube-API aktivieren. Sie können ihn jedoch auch anhand der folgenden Anmeldeinformationen überprüfen.
DAG-Definition, die in Bigquery gespeichert werden soll
Unten finden Sie den DAG-Code, der die Youtubeapi-Klasse verwendet, um die Youtube-API aufzurufen und die Daten in Bigquery zu speichern.
dag.py
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
from google.cloud import bigquery
from lib import youtubelib
start = DummyOperator(task_id='start')
GCP_PROJECT =Projektname
BQ_YOUTUBE_DATASET =Datensatzname.Tabellenname
YOUTUBE＿CHANNEL＿ID =Tokai On Air Channel ID('UCutJqz56653xV2wwSvut_hQ')
YOUTUBE_API_KEY =Youtube API-Schlüssel
# see https://cloud-textbook.com/69/#_start_dateschedule_intervalcatchup
default_args = {
'owner': 'airflow',
'start_date': datetime(2020, 5, 18),
'retries': 0,
'max_active_runs': 1,
}
schedule_interval = '0 17 * * *'
# define dag
dag = DAG('toukai_trends', default_args=default_args,
schedule_interval=schedule_interval, catchup=False)
youtubeapi = youtubelib.Youtubeapi(channel_id=YOUTUBE_CHANNEL_ID, api_key=YOUTUBE_API_KEY)
def pull_youtube_statistics_api(ds, **kwargs):
statistics = youtubeapi.get_statistics_data()
dt = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print('dt', dt)
viewcount = statistics['viewCount']
print('viewCount', viewcount)
subscribercount = statistics['subscriberCount']
print('subscriberCount', subscribercount)
videocount = statistics['videoCount']
print('videoCount', videocount)
bigquery_client = bigquery.Client()
query = '''
INSERT
`{0}.{1}`
SELECT
CAST("{2}" AS timestamp) AS datetime,
CAST("{3}" AS INT64) AS viewcount,
CAST("{4}" AS INT64) as subscribercount,
CAST("{5}" AS INT64) as videocount
'''.format(GCP_PROJECT, BQ_YOUTUBE_DATASET, str(dt), str(viewcount), str(subscribercount), str(videocount))
rows = bigquery_client.query(query).result()
return 'ok'
job_transactiondetail_puller = PythonOperator(
task_id='pull_youtube_statistics_api',
provide_context=True,
python_callable=pull_youtube_statistics_api,
dag=dag,
)
Es gibt verschiedene komplizierte Punkte in Bezug auf die reguläre Ausführung von Composer (Dag-Teil definieren), so dass die reguläre Ausführung von gründlicher Vergleich / gründliche Erklärung des Cloud-Dienstes (Version 2020) Bitte überprüfen Sie die verwirrende Story-Ausgabe (Startdatum, Zeitplanintervall, Aufholjagd usw.).
Laden Sie die DAG-Definition in Cloud Composer hoch
Laden Sie dag.py in die Cloud Shell hoch und führen Sie die folgenden Schritte in der Shell aus:
gcloud composer environments storage dags import --environment=Komponistenname
(toukai-onair-composer) --location=asia-northeast1 --source=./dag.py
Wenn Sie es ausführen, wird eine Datei in GCS erstellt und dort wird auch youtubelib.py abgelegt.
Funktionsprüfung
Entspricht der Prüfung des Ausführungsstatus der Cloud Composer-DAG in Verfahren zum Erstellen eines Workflows zum Bearbeiten von BigQuery-Tabellen in GCP Cloud Composer Wenn Sie es überprüfen, gibt es kein Problem.
Sie können das Ausführungsergebnisprotokoll auch in den Ansichtsprotokollen in Airflow überprüfen.
Ein Gefühl von Preis
Ich habe es bemerkt, als ich es ausprobiert habe, aber es ist immer noch teuer. Es war ein schmerzhaftes Level für mein Taschengeld, also habe ich lange aufgehört, es zu betreiben ... (In diesem Fall ist die in Youtube Premium registrierte Lebensqualität hoch ...) Ein Gefühl für den Preis finden Sie in der "Price Edition" von Gründlicher Vergleich und ausführliche Erläuterung der Cloud-Dienste (Version 2020).
Zusammenfassung
Ich habe versucht, es zu machen, aber es schien schwierig, es persönlich zu benutzen, weil es zu teuer war ...
Das nächste Mal frage ich mich, ob es durch die regelmäßige Ausführung von Cloud Finction billiger sein kann. Deshalb denke ich darüber nach, es zu erstellen.
Lass uns ein gutes Youtube-Leben haben! !!
Recommended Posts