Como extrair dados do Google Analytics para o Cloud Storage pelo Dataproc com Python

Nesta postagem, falaremos como exportar os dados que existam no Google Analytics para um bucket do Cloud Storage.

O que é Google Analytics?

É uma plataforma do Google que permite o monitoramento de trafego de um site ou de uma aplicação, através de relatórios o G.A. pode demonstrar o comportamento de um grupo de usuários.

O que é Cloud Storage?

Fornece, a nível mundial, um armazenamento de objetos altamente durável que escalona para exabytes de dados. É possível acessar dados instantaneamente de qualquer classe de armazenamento, integrar o armazenamento aos seus aplicativos com uma API exclusiva unificada e otimizar o preço e desempenho de forma fácil.

O que é Dataproc?

É a plataforma de big data totalmente gerenciado, para processar grandes quantidades de dados com rapidez, de forma econômica e em grande escala. Usando ferramentas de código aberto como o Apache Spark, o Apache Hive, o Apache Hadoop e o Apache Pig combinadas à escalabilidade dinâmica do Compute Engine e ao armazenamento escalável do Cloud Storage, o Dataproc oferece às equipes analíticas os mecanismos e a elasticidade para executar análises na escala de petabytes por uma fração do custo dos clusters locais tradicional, além de ser facilmente incorporado a outros serviços do Google Cloud Platform (GCP).

Passo a Passo

1. O primeiro passo a fazer é o login em sua conta do GCP, para este tutorial você vai precisar de uma conta que possua um projeto.

2. Vá até a barra superior e clique no ícone do Cloud Shell, igual a imagem abaixo.

3. Agora temos que preparar as variáveis de ambiente que usaremos nos outros passos:

a) Para definir a variável de região usaremos o código abaixo:

export REGION=us-central1

b) Para definir a variável de Zona usaremos o código abaixo:

export ZONE=us-central1-a

4. Nesse passo vamos definir o projeto a ser usado dentro do cloud shell, no campo [PROJECT_ID], insira o id do seu projeto:

gcloud config set project [PROJECT_ID]

5. Nesse passo vamos definir a Zona a ser usada dentro do cloud shell com base na variável que definimos antes:

gcloud config set compute/zone $ZONE

6. Nesse passo vamos definir a Região a ser usada dentro do cloud shell com base na variável que definimos antes:

gcloud config set compute/region $REGION

7. Após isso vamos definir a variável PROJECT com o id do projeto:

export PROJECT=$(gcloud info --format='value(config.project)')

8. Com o código abaixo a api do dataproc será habilitada, caso já esteja pode desconsiderar:

gcloud services enable dataproc.googleapis.com sqladmin.googleapis.com

9. Criaremos o bucket do Cloud Storage em que iremos exportar nossa tabela:

gsutil mb -l us-central1 gs://[PROJECT_ID]-arquivos

10. Caso não exista, crie um cluster do dataproc com o comando abaixo:

gcloud dataproc clusters create meu-cluster \
 --image-version 1.3 \
 --region $REGION \
 --subnet default \
 --master-machine-type n1-standard-1 \
 --master-boot-disk-size 500 \
 --num-workers 2 \
 --worker-machine-type n1-standard-1 \
 --worker-boot-disk-size 500

Observações:

A propriedade “subnet” faz referência a sub-rede que usaremos no cluster.

A propriedade “master-machine-type” permite selecionar o tipo de máquina a ser usado pelo master.

A propriedade “worker-machine-type” permite selecionar o tipo de máquina a ser usado pelos workes.

11. Vá ate o menu do Compute Engine, seleciona a maquina Master de seu cluster e execute SSH:

12. Execute os comandos abaixo, nele :

#Criação e ativação de um ambiente chamado gaToPandas
virtualenv gaToPandas
source gaToPandas/bin/activate
#Instalação de libs necessárias
pip install --upgrade google-api-python-client
pip install oauth2client
pip install pandas
pip install pyarrow
pip install gcsfs

13. Pelo console, no menu do IAM, selecione contas de serviço:

14. Selecione uma conta de serviço que tenha permissão no bucket, clique em Criar chave:

15. Gere um arquivo Json, esse arquivo será baixado na sua máquina, renomeie esse arquivo para "client_secrets.json" e jogue na sua home dentro da máquina master do Dataproc:

16. Vá até o menu APIs e serviços:

17. Pesquise por "Google Analytics Reporting API" e ative:

18. Vá até o portal do Google Analytics e vá até o menu "Administrador":

19. Selecione "Gerenciamento de usuários da propriedade":

20. Selecione Adicionar usuários:

21. Digite o email da conta de serviço que a chave foi gerada no passo 13, habilite as permissões necessárias e clique em adicionar.

22. De volta ao SSH do Dataproc, crie um novo arquivo chamado GaToDataproc.py:

nano GaToDataproc.py

23. Coloque este código no arquivo, substitua os locais informados:

#Libs necessárias
import pandas as pd
from apiclient.discovery import build
from oauth2client.service_account import ServiceAccountCredentials


SCOPES = ['https://www.googleapis.com/auth/analytics.readonly']
#Arquivo de credencial
KEY_FILE_LOCATION = 'client_secrets.json'
#Id de visualização do GA, substitua pelo seu ID
VIEW_ID = '[SEU_ID]'

#Dimensão e métrica a ser extraída do GA, , substitua pela desejada.
DIMENSIONS = ['ga:source','ga:medium']
METRICS = ['ga:users','ga:sessions']

def initialize_analyticsreporting():
  credentials = ServiceAccountCredentials.from_json_keyfile_name(
      KEY_FILE_LOCATION, SCOPES)

  # Build the service object.
  analytics = build('analyticsreporting', 'v4', credentials=credentials)

  return analytics


def get_report(analytics):
  return analytics.reports().batchGet(
      body={
        'reportRequests': [
        {
          'viewId': VIEW_ID,
          'dateRanges': [{'startDate': '7daysAgo', 'endDate': 'today'}],
          'metrics': [{'expression':i} for i in METRICS],
          'dimensions': [{'name':j} for j in DIMENSIONS]
        }]
      }
  ).execute()


def convert_to_dataframe(response):

  for report in response.get('reports', []):
    columnHeader = report.get('columnHeader', {})
    dimensionHeaders = columnHeader.get('dimensions', [])
    metricHeaders = [i.get('name',{}) for i in columnHeader.get('metricHeader', {}).get('metricHeaderEntries', [])]
    finalRows = []


    for row in report.get('data', {}).get('rows', []):
      dimensions = row.get('dimensions', [])
      metrics = row.get('metrics', [])[0].get('values', {})
      rowObject = {}

      for header, dimension in zip(dimensionHeaders, dimensions):
        rowObject[header] = dimension


      for metricHeader, metric in zip(metricHeaders, metrics):
        rowObject[metricHeader] = metric

      finalRows.append(rowObject)


  dataFrameFormat = pd.DataFrame(finalRows)
  return dataFrameFormat

#pasta em que o arquivo será gerado, substitua para colocar seu diretório
def export_to_storage(df):
  df.to_parquet('gs://[PROJECT_ID]-arquivos/teste.parquet')


def main():
  analytics = initialize_analyticsreporting()
  response = get_report(analytics)
  df = convert_to_dataframe(response)   #df = pandas dataframe
  export_to_storage(df)                 
  print(df)

if __name__ == '__main__':
  main()

24. Execute seu código: