ื ืืืื ืืขืืขืจ. ืืืจ ืืืืื ืื ืืืืขืจืืขืฆืื ื ืคืื ืื ืืขืฆื ืืืื ืคืื ืืขื ืึทืจืืืงื, ืฆืืืขืืจืืื ืกืคึผืึทืกืืคืืงืื ืคึฟืึทืจ ืกืืืืขื ืื ืคืื ืืขื ืงืืจืก. . ืืืจ ืงืขื ื ืืืืขื ืขื ืืขื ืขืจืฉืืขืจ ืืืื .
ืึทืคึผืึทืืฉื ืฉืืจืึทื ืืื ืืึทืืึทืคืืึธืื ืคึฟืึทืจ ืคืึทืงืืืฉ-ืฆืืื ืคึผืืืคึผืืืื ื
ืืึทืฉืืขืืืงื Google ืงืืึธืื
ืืึทืืขืจืงืื ื: ืืื ืืขืืืืื ื Google Cloud Shell ืฆื ืืืืคื ืื ืจืขืจื - ืืื ืืข ืืื ืึทืจืืืกืืขืื ืื ืื ืืึธื ืืึทืื ืืืืึทื ืืื ืืื ืืขืืืขื ืงืึธื ืคืืืงื ืืื ืื ืจืขืจื - ืืื ืืข ืืื Python 3. Google Cloud Shell ื ืืฆื ืคึผืืืืึธื 2, ืืืึธืก ืืื ืืขืจ ืงืึธื ืกืืกืืขื ื ืืื ืึทืคึผืึทืืฉื ืฉืืจืึทื.
ืฆื ืึธื ืืืืื ืื ืจืขืจื - ืืื ืืข, ืืืจ ืืึทืจืคึฟื ืฆื ืืจืึธืื ืึท ืืืกื ืืื ืื ืกืขืืืื ืืก. ืคึฟืึทืจ ืื ืคืื ืืืจ ืืืึธืก ืืึธืื ื ืืฉื ืืขืืืืื ื GCP ืคืจืืขืจ, ืืืจ ืืึทืจืคึฟื ืฆื ื ืึธืืคืึธืืื ืื ืคืืืืขื ืืข 6 ืกืืขืคึผืก ืึทืืืืืื ื ืืื ืืขื .
ื ืึธื ืืขื, ืืืจ ืืึทืจืคึฟื ืฆื ืฆืืคึฟืขืืืงืขืจ ืืื ืืืขืจ ืกืงืจืืคึผืก ืฆื Google ืงืืึธืื ืกืืึธืจืืืืฉ ืืื ื ืึธืืืึทืื ืืื ืฆื ืืื ืืืขืจ Google ืงืืึธืื ืฉืขื. ืืคึผืืึธืึทืืื ื ืฆื ืืืึธืืงื ืกืืึธืจืืืืฉ ืืื ืืึทื ืฅ ืืจืืืืืึทื (ืึท ืืึทืฉืจืืึทืืื ื ืงืขื ืขื ืืืื ืืขืคึฟืื ืขื ). ืฆื ื ืึธืืืึทืื ืืื ืืืขืจ ืืขืงืขืก, ืืืจ ืงืขื ืขื ืขืคึฟืขื ืขื Google Cloud Shel ืคึฟืื ืื ืืืฉืืจ ืืืจื ืืขืื ืึท ืงืืืง ืืืืฃ ืืขืจ ืขืจืฉืืขืจ ืืืืื ืืืืฃ ืื ืืื ืงืก ืืื ืคืืืืจืข 2 ืืื ืื.
ืคืืืืจืข ืงืกื ืืืงืก
ืื ืงืึทืืึทื ืื ืืืึธืก ืืืจ ืืึทืจืคึฟื ืฆื ื ืึธืืืึทืื ืื ืืขืงืขืก ืืื ืื ืกืืึทืืืจื ืื ืคืืจืืื ืื ืืืืืจืขืจืื ืืขื ืขื ืืืกืืขื ืืื ืื.
# Copy file from cloud storage
gsutil cp gs://<YOUR-BUCKET>/ * .
sudo pip install apache-beam[gcp] oauth2client==3.0.0
sudo pip install -U pip
sudo pip install Faker==1.0.2
# Environment variables
BUCKET=<YOUR-BUCKET>
PROJECT=<YOUR-PROJECT>ืงืจืืืืืื ื ืืื ืืืขืจ ืืึทืืึทืืืืก ืืื ืืืฉ
ืึทืืึธื ืืืจ ืืึธืื ืืืจืืืขืงืึธืื ืึทืืข ืื ืกืขืืึทืคึผ ืฉืืึทืืืช ืกืืขืคึผืก, ืืขืจ ืืืืึทืืขืจ ืืึทื ืืืจ ืืึทืจืคึฟื ืฆื ืืึธื ืืื ืฉืึทืคึฟื ืึท ืืึทืืึทืกืขื ืืื ืืืฉ ืืื BigQuery. ืขืก ืืขื ืขื ืขืืืขืืข ืืืขืื ืฆื ืืึธื ืืึธืก, ืึธืืขืจ ืื ืกืืืคึผืืึทืกื ืืื ืฆื ื ืืฆื ืื Google ืงืืึธืื ืงืึทื ืกืึธืื ืืืจื ืขืจืฉืืขืจ ืงืจืืืืืื ื ืึท ืืึทืืึทืกืขื. ืืืจ ืงืขื ืขื ื ืึธืืืืื ืื ืกืืขืคึผืก ืืื ืื ืฆื ืฉืึทืคึฟื ืึท ืืืฉ ืืื ืึท ืกืืฉืขืืึท. ืืื ืืืขืจ ืืืฉ ืืืขื ืืึธืื 7 ืฉืคืืืื, ืงืึธืจืึทืกืคึผืึทื ืืื ื ืฆื ืื ืงืึทืืคึผืึธืื ืึทื ืฅ ืคืื ืืขืืขืจ ืืึทื ืืฆืขืจ ืงืืึธืฅ. ืคึฟืึทืจ ืงืึทื ืืืื ืืึทื ืก, ืืืจ ืืืขืื ืืขืคืื ืืจื ืึทืืข ืฉืคืืืื ืืื ืกืืจืื ืืก, ืึทืืืฅ ืคึฟืึทืจ ืื ืืืืขืืึธืงืึทื ืืืึทืืขืืืืืง, ืืื ื ืขืืขื ืืื ืืืื ืื ืืืขืจืืึทืืึทืื ืืืึธืก ืืืจ ืืืฉืขื ืขืจืืืืึทื ืคืจืืขืจ. ืืขืจ ืืืืกืืืื ืคืื ืืื ืืืขืจ ืืืฉ ืืึธื ืงืืงื ืืื ืืื ืคืืืืจืข 3.
ืคืืืืจืข 3. ืืืฉ ืืืืกืืืื
ืืจืืืกืืขืื ืืึทื ืืฆืขืจ ืงืืึธืฅ ืืึทืื
ืคึผืื / ืกืื ืืื ืึท ืงืจืืืืฉ ืงืึธืืคึผืึธื ืขื ื ืคืื ืืื ืืืขืจ ืจืขืจื - ืืื ืืข ืืืืึทื ืขืก ืึทืืึทืื ืงืืืคื ืคืจืืึท ืึทืคึผืืึทืงืืืฉืึทื ื ืฆื ืืืขืจืืขืื ืืื ืืขืืขืจ ืื ืืขืจืขืจ. ืืื ืืึทืืื ืืขืจ, ืขืก ืึทืจืืขื ืืื ืึท ืื ืืขืจืืืืืขืจื ืืืึธืก ืึทืืึทืื ืืื ืื ืฆื ืฉืืงื ืืื ืืึทืงืืืขื ืึทืจืืืงืืขื ืฆืืืืฉื ืึทืคึผืืึทืงืืืฉืึทื ื. ืืขืจ ืขืจืฉืืขืจ ืืึทื ืืืจ ืืึทืจืคึฟื ืฆื ืืึธื ืืื ืฉืึทืคึฟื ืึท ืืขืืข. ืคืฉืื ืืืื ืฆื Pub / Sub ืืื ืื ืงืึทื ืกืึธืื ืืื ืืื CREATE TOPIC.
ืืขืจ ืงืึธื ืืื ืื ืจืืคื ืืื ืืืขืจ ืฉืจืืคื ืฆื ืืืฉืขื ืขืจืืื ืื ืงืืึธืฅ ืืึทืื ืืืคืืื ื ืืืืื ืืื ืืขืืึธืื ืงืึทื ืขืงืฅ ืืื ืกืขื ืื ืื ืืึธืืก ืฆื ืคึผืื / ืกืื. ืืขืจ ืืืืื ืืึทื ืืืจ ืืึทืจืคึฟื ืฆื ืืึธื ืืื ืฉืึทืคึฟื ืึท ืืืืคืขืฅ PublisherClient, ืกืคึผืขืฆืืคืืฆืืจื ืืขื ืืจื ืฆื ืืขืจ ืืขืืข ื ืืฆื ืืขื ืืืคึฟื topic_path ืืื ืจืืคื ืื ืคึฟืื ืงืฆืืข publish ั topic_path ืืื ืืึทืื. ืืืืข ืืึธื ืึทื ืืืจ ืึทืจืืึทื ืคืืจ generate_log_line ืคืื ืืื ืืืขืจ ืฉืจืืคื stream_logs, ืึทืืื ืืึทืื ืืืืขืจ ืึทื ืื ืืขืงืขืก ืืขื ืขื ืืื ืืขืจ ืืขืืืืงืขืจ ืืขืงืข, ืึทื ืืขืจืฉ ืืืจ ืืืขื ืืึทืงืืืขื ืึท ืึทืจืืึทื ืคืืจ ืืขืืช. ืืืจ ืงืขื ืขื ืืขืจืืืขืจ ืืืืคื ืืขื ืืืจื ืืื ืืืขืจ Google ืงืึทื ืกืึธืื ื ืืฆื:
python publish.pyfrom stream_logs import generate_log_line
import logging
from google.cloud import pubsub_v1
import random
import time
PROJECT_ID="user-logs-237110"
TOPIC = "userlogs"
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(PROJECT_ID, TOPIC)
def publish(publisher, topic, message):
data = message.encode('utf-8')
return publisher.publish(topic_path, data = data)
def callback(message_future):
# When timeout is unspecified, the exception method waits indefinitely.
if message_future.exception(timeout=30):
print('Publishing message on {} threw an Exception {}.'.format(
topic_name, message_future.exception()))
else:
print(message_future.result())
if __name__ == '__main__':
while True:
line = generate_log_line()
print(line)
message_future = publish(publisher, topic_path, line)
message_future.add_done_callback(callback)
sleep_time = random.choice(range(1, 3, 1))
time.sleep(sleep_time)ืืื ืืึทืื ืืื ืื ืืขืงืข ืืืืคื, ืืืจ ืงืขื ืขื ืืขื ืื ืจืขืืืืืึทื ืคืื ืื ืงืืึธืฅ ืืึทืื ืฆื ืื ืงืึทื ืกืึธืื, ืืื ืืขืืืืื ืืื ืื ืคืืืืจ ืืื ืื. ืืขืจ ืฉืจืืคื ืืืขื ืึทืจืืขืื ืึทืืื ืืึทื ื ืืื ืืืจ ืืึธื ื ืื ื ืืฆื ืงืืจื + Cืฆื ืคืึทืจืขื ืืืงื ืขืก.
ืคืืืืจืข 4. ืจืขืืืืืึทื publish_logs.py
ืฉืจืืืื ืืื ืืืขืจ ืจืขืจื - ืืื ืืข ืงืึธื
ืืืฆื ืึทื ืืืจ ืืึธืื ืึทืืฅ ืฆืืืขืืจืืื, ืืืจ ืงืขื ืขื ืึธื ืืืืื ืืขื ืฉืคึผืึทืก ืืืื - ืงืึธืืืจืื ื ืคืื ืืื ืืืขืจ ืจืขืจื - ืืื ืืข ืืื ืฉืืจืึทื ืืื ืคึผืืืืึธื. ืฆื ืฉืึทืคึฟื ืึท ืฉืืจืึทื ืจืขืจื - ืืื ืืข, ืืืจ ืืึทืจืคึฟื ืฆื ืฉืึทืคึฟื ืึท ืจืขืจื - ืืื ืืข ืืืืคืขืฅ (ืคึผ). ืึทืืึธื ืืืจ ืืึธืื ืืืฉืืคื ืึท ืจืขืจื - ืืื ืืข ืืืืคืขืฅ, ืืืจ ืงืขื ืขื ืฆืืืืืื ืงืืืคื ืคืึทื ืืงืฉืึทื ื ืืืื ืขืจ ื ืึธื ืืขื ืื ืืขืจื ื ืืฆื ืื ืึธืคึผืขืจืึทืืึธืจ pipe (|). ืืื ืึทืืืขืืืื, ืื ืืืึธืจืงืคืืึธืื ืงืืงื ืืื ืื ืืืื ืืื ืื.
[Final Output PCollection] = ([Initial Input PCollection] | [First Transform]
| [Second Transform]
| [Third Transform])
ืืื ืืื ืืืขืจ ืงืึธื, ืืืจ ืืืขืื ืืึทืื ืฆืืืื ืื ืื ืคืึทื ืืงืฉืึทื ื. ืคึฟืื ืงืฆืืข regex_clean, ืืืึธืก ืกืงืึทื ื ืื ืืึทืื ืืื ืจืืืจืืืื ืื ืงืึธืจืึทืกืคึผืึทื ืืื ื ืจืืืขืจื ืืืืืจื ืืืืฃ ืื PATTERNS ืจืฉืืื ื ืืฆื ืื ืคึฟืื ืงืฆืืข re.search. ืื ืคึฟืื ืงืฆืืข ืงืขืจื ืึท ืงืึธืืข ืืคืืขืฉืืืื ืฉืืจืืงื. ืืืื ืืืจ ืืขื ื ื ืืฉื ืึท ืจืขืืืืขืจ ืืืืกืืจืืง ืืืืื, ืืื ืจืขืงืึธืืขื ืืืจื ืฆื ืงืึธื ืืจืึธืืืจื ืืขื ืืื ืคืืจ ืืื ืึท ื ืึธืืขืคึผืึทื ืฆื ืงืึธื ืืจืึธืืืจื ืืขื ืงืึธื. ื ืึธื ืืขื, ืืืจ ืืขืคืื ืืจื ืึท ืื ืื ืคึผืึทืจืืึธ ืคึฟืื ืงืฆืืข ืืขืจืืคื ืฉืคึผืึทืืื, ืืืึธืก ืืื ืึท ืืืขืจืืืืฉืึทื ืคืื ืื ืฉืืจืึทื ืืืขืจืืึทืื ืคึฟืึทืจ ืคึผืึทืจืึทืืขื ืคึผืจืึทืกืขืกืื ื. ืืื ืคึผืืืืึธื, ืืึธืก ืืื ืืขืืื ืืื ืึท ืกืคึผืขืฆืืขื ืืืขื - ืืืจ ืืืื ืืึทืื ืึท ืงืืึทืก ืืืึธืก ืื ืืขืจืึทืฅ ืคืื ืื DoFn Beam ืงืืึทืก. ืื ืกืคึผืืื ืคึฟืื ืงืฆืืข ื ืขืื ืื ืคึผืึทืจืกื ืจืืืขืจื ืคืื ืื ืคืจืืขืจืืืงืข ืคึฟืื ืงืฆืืข ืืื ืงืขืจื ืึท ืจืฉืืื ืคืื ืืืงืฉืึทื ืขืจืื ืืื ืฉืืืกืืขื ืงืึธืจืึทืกืคึผืึทื ืืื ื ืฆื ืื ืืืึทื ื ืขืืขื ืืื ืืื ืืืขืจ BigQuery ืืืฉ. ืขืก ืืื ืขืคึผืขืก ืฆื ืืึธื ืืืขืื ืืขื ืคึฟืื ืงืฆืืข: ืืื ืืื ืฆื ืึทืจืืึทื ืคืืจ datetime ืืื ืึท ืคึฟืื ืงืฆืืข ืฆื ืืึทืื ืขืก ืึทืจืืขื. ืืื ืืื ืืขืืืขื ืืขืืื ื ืึท ืึทืจืืึทื ืคืืจ ืืขืืช ืืื ืื ืึธื ืืืื ืคืื ืืขืจ ืืขืงืข, ืืืึธืก ืืื ืืขืืืขื ืืฉืืื ืข. ืืขืจ ืจืฉืืื ืืื ืืขืจื ืึธื ืืืจืืืขืืื ืืขื ืฆื ืื ืคึฟืื ืงืฆืืข WriteToBigQuery, ืืืึธืก ืคืฉืื ืืืกืืฃ ืืื ืืืขืจ ืืึทืื ืฆื ืื ืืืฉ. ืืขืจ ืงืึธื ืคึฟืึทืจ ืืึทืืฉ ืืึทืืึทืคืืึธืื ืืืฉืึธื ืืื ืกืืจืืืื ื ืืึทืืึทืคืืึธืื ืืืฉืึธื ืืื ืืขืืขืื ืืื ืื. ืืขืจ ืืืืื ืืืืืง ืฆืืืืฉื ืคึผืขืงื ืืื ืกืืจืืืื ื ืงืึธื ืืื ืึทื ืืื ืคึผืขืงื ืืืจ ืืืืขื ืขื ืื CSV ืคึฟืื src_pathื ืืฆื ืื ืคึฟืื ืงืฆืืข ReadFromText ืคืื ืืึทื.
ืืึทืืฉ ืืึทืืึทืคืืึธืื ืึทืจืืขื (ืืึทืืฉ ืคึผืจืึทืกืขืกืื ื)
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import bigquery
import re
import logging
import sys
PROJECT='user-logs-237110'
schema = 'remote_addr:STRING, timelocal:STRING, request_type:STRING, status:STRING, body_bytes_sent:STRING, http_referer:STRING, http_user_agent:STRING'
src_path = "user_log_fileC.txt"
def regex_clean(data):
PATTERNS = [r'(^S+.[S+.]+S+)s',r'(?<=[).+?(?=])',
r'"(S+)s(S+)s*(S*)"',r's(d+)s',r"(?<=[).d+(?=])",
r'"[A-Z][a-z]+', r'"(http|https)://[a-z]+.[a-z]+.[a-z]+']
result = []
for match in PATTERNS:
try:
reg_match = re.search(match, data).group()
if reg_match:
result.append(reg_match)
else:
result.append(" ")
except:
print("There was an error with the regex search")
result = [x.strip() for x in result]
result = [x.replace('"', "") for x in result]
res = ','.join(result)
return res
class Split(beam.DoFn):
def process(self, element):
from datetime import datetime
element = element.split(",")
d = datetime.strptime(element[1], "%d/%b/%Y:%H:%M:%S")
date_string = d.strftime("%Y-%m-%d %H:%M:%S")
return [{
'remote_addr': element[0],
'timelocal': date_string,
'request_type': element[2],
'status': element[3],
'body_bytes_sent': element[4],
'http_referer': element[5],
'http_user_agent': element[6]
}]
def main():
p = beam.Pipeline(options=PipelineOptions())
(p
| 'ReadData' >> beam.io.textio.ReadFromText(src_path)
| "clean address" >> beam.Map(regex_clean)
| 'ParseCSV' >> beam.ParDo(Split())
| 'WriteToBigQuery' >> beam.io.WriteToBigQuery('{0}:userlogs.logdata'.format(PROJECT), schema=schema,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
)
p.run()
if __name__ == '__main__':
logger = logging.getLogger().setLevel(logging.INFO)
main()ืกืืจืืืื ื DataFlow ืึทืจืืขื (ืกืืจืื ืคึผืจืึทืกืขืกืื ื)
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import pubsub_v1
from google.cloud import bigquery
import apache_beam as beam
import logging
import argparse
import sys
import re
PROJECT="user-logs-237110"
schema = 'remote_addr:STRING, timelocal:STRING, request_type:STRING, status:STRING, body_bytes_sent:STRING, http_referer:STRING, http_user_agent:STRING'
TOPIC = "projects/user-logs-237110/topics/userlogs"
def regex_clean(data):
PATTERNS = [r'(^S+.[S+.]+S+)s',r'(?<=[).+?(?=])',
r'"(S+)s(S+)s*(S*)"',r's(d+)s',r"(?<=[).d+(?=])",
r'"[A-Z][a-z]+', r'"(http|https)://[a-z]+.[a-z]+.[a-z]+']
result = []
for match in PATTERNS:
try:
reg_match = re.search(match, data).group()
if reg_match:
result.append(reg_match)
else:
result.append(" ")
except:
print("There was an error with the regex search")
result = [x.strip() for x in result]
result = [x.replace('"', "") for x in result]
res = ','.join(result)
return res
class Split(beam.DoFn):
def process(self, element):
from datetime import datetime
element = element.split(",")
d = datetime.strptime(element[1], "%d/%b/%Y:%H:%M:%S")
date_string = d.strftime("%Y-%m-%d %H:%M:%S")
return [{
'remote_addr': element[0],
'timelocal': date_string,
'request_type': element[2],
'body_bytes_sent': element[3],
'status': element[4],
'http_referer': element[5],
'http_user_agent': element[6]
}]
def main(argv=None):
parser = argparse.ArgumentParser()
parser.add_argument("--input_topic")
parser.add_argument("--output")
known_args = parser.parse_known_args(argv)
p = beam.Pipeline(options=PipelineOptions())
(p
| 'ReadData' >> beam.io.ReadFromPubSub(topic=TOPIC).with_output_types(bytes)
| "Decode" >> beam.Map(lambda x: x.decode('utf-8'))
| "Clean Data" >> beam.Map(regex_clean)
| 'ParseCSV' >> beam.ParDo(Split())
| 'WriteToBigQuery' >> beam.io.WriteToBigQuery('{0}:userlogs.logdata'.format(PROJECT), schema=schema,
write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND)
)
result = p.run()
result.wait_until_finish()
if __name__ == '__main__':
logger = logging.getLogger().setLevel(logging.INFO)
main()
ืกืืึทืจืืื ื ืื ืงืึทื ืืืืืขืจ
ืืืจ ืงืขื ืขื ืืืืคื ืื ืจืขืจื - ืืื ืืข ืืื ืขืืืขืืข ืคืึทืจืฉืืืขื ืข ืืืขืื. ืืืื ืืืจ ืืืขืื, ืืืจ ืงืขื ื ืึธืจ ืืืืคื ืขืก ืืึธืืงืึทืื ืคึฟืื ืึท ืืืึธืงืืึทื ืืฉืขืช ืืึธืืื ื ืืื GCP ืจืืืึธืืืื.
python -m main_pipeline_stream.py
--input_topic "projects/user-logs-237110/topics/userlogs"
--streamingืึธืืขืจ, ืืืจ ืืืขืื ืืืืคื ืขืก ืืื DataFlow. ืืืจ ืงืขื ืขื ืืึธื ืืึธืก ื ืืฆื ืื ืืื ืื ืืึทืคึฟืขื ืืืจื ืืึทืฉืืขืืืงื ืื ืคืืืืขื ืืข ืคืืจืืื ืื ืคึผืึทืจืึทืืขืืขืจืก.
project- ID ืคืื ืืืื GCP ืคึผืจืืืขืงื.runnerืืื ืึท ืจืขืจื - ืืื ืืข ืืืึธืก ืืืขื ืึทื ืึทืืืื ืืืื ืคึผืจืึธืืจืึทื ืืื ืืืืขื ืืืื ืจืขืจื - ืืื ืืข. ืฆื ืืืืคื ืืื ืื ืืืึธืืงื, ืืืจ ืืืื ืกืคึผืขืฆืืคืืฆืืจื ืึท DataflowRunner.staging_location- ืืขืจ ืืจื ืฆื ืื ืงืืึธืื ืืึทืืึทืคืืึธืื ืืืึธืืงื ืกืืึธืจืืืืฉ ืคึฟืึทืจ ืื ืืขืงืกืื ื ืงืึธื ืคึผืึทืงืึทืืืฉืึทื ืืืจืฃ ืืืจื ืื ืคึผืจืึทืกืขืกืขืจื ืืืึธืก ืืืจืืคืืจื ืื ืึทืจืืขื.temp_location- ืืจื ืฆื ืื ืงืืึธืื ืืึทืืึทืคืืึธืื ืืืึธืืงื ืกืืึธืจืืืืฉ ืคึฟืึทืจ ืกืืึธืจืื ื ืฆืืึทืืืืืึทืืืง ืึทืจืืขื ืืขืงืขืก ืืืฉืืคื ืืฉืขืช ืื ืจืขืจื - ืืื ืืข ืืื ืคืืืกื ืืืง.streaming
python main_pipeline_stream.py
--runner DataFlow
--project $PROJECT
--temp_location $BUCKET/tmp
--staging_location $BUCKET/staging
--streaming
ืืฉืขืช ืืขื ืืึทืคึฟืขื ืืื ืคืืืกื ืืืง, ืืืจ ืงืขื ืขื ืืืื ืฆื ืื DataFlow ืงืืืืื ืืื ืื Google ืงืึทื ืกืึธืื ืืื ืืขื ืืื ืืืขืจ ืจืขืจื - ืืื ืืข. ืืืขื ืืืจ ืืจืืงื ืืืืฃ ืื ืจืขืจื - ืืื ืืข, ืืืจ ืืึธื ืืขื ืขืคึผืขืก ืขื ืืขื ืฆื ืคืืืืจืข 4. ืคึฟืึทืจ ืืืืึทืืื ื ืฆืืืขืงื, ืขืก ืงืขื ืืืื ืืืืขืจ ื ืืฆืืง ืฆื ืืืื ืฆื ืืึธืืก ืืื ืืึทื ืกืืึทืงืืจืืืืขืจ ืฆื ืืขื ืืืืืืื ืืึธืืก. ืืึธืก ืืึธื ืืขืืึธืืคึฟื ืืืจ ืกืึธืืืืข ืจืขืจื - ืืื ืืข ืืฉืื ืืื ืึท ื ืืืขืจ ืคืื ืงืึทืกืขืก.
ืคืืืืจืข 4: ืฉืืจืึทื ืงืึทื ืืืืืขืจ
ืึทืงืกืขืก ืืื ืืืขืจ ืืึทืื ืืื BigQuery
ืึทืืื, ืืืจ ืืึธื ืฉืืื ืืึธืื ืึท ืจืขืจื - ืืื ืืข ืคืืืกื ืืืง ืืื ืืึทืื ืคืืึธืืื ื ืืื ืืื ืืืขืจ ืืืฉ. ืฆื ืคึผืจืืืืจื ืืึธืก, ืืืจ ืงืขื ืขื ืืืื ืฆื BigQuery ืืื ืงืืง ืืื ืื ืืึทืื. ื ืึธื ื ืืฆื ืื ืืึทืคึฟืขื ืืื ืื ืืืจ ืืึธื ืืขื ืื ืขืจืฉืืขืจ ืืืกื ืจืึธืื ืคืื ืื ืืึทืืึทืกืขื. ืืืฆื ืึทื ืืืจ ืืึธืื ืื ืืึทืื ืกืืึธืจื ืืื BigQuery, ืืืจ ืงืขื ืขื ืึธื ืคืืจื ืืืืึทืืขืจ ืึทื ืึทืืืกืืก, ืืื ืืืื ืืืืื ืื ืืึทืื ืืื ืืืจืื ืืื ืึธื ืืืืื ืขื ืืคึฟืขืจื ืืขืฉืขืคื ืคึฟืจืืื.
SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10;
ืคืืืืจืข 5: ืืืืงืืืขืจื
ืกืึธืฃ
ืืืจ ืืึธืคื ืึทื ืืขื ืคึผืึธืกืื ืกืขืจืืืขืก ืืื ืึท ื ืืฆืืง ืืืืฉืคึผืื ืคืื ืงืจืืืืืื ื ืึท ืกืืจืืืื ื ืืึทืื ืจืขืจื - ืืื ืืข, ืืื ืืขืืื ื ืืื ืฆื ืืขืคึฟืื ืขื ืืืขืื ืฆื ืืึทืื ืืึทืื ืืขืจ ืฆืืืจืืืืขื. ืกืืึธืจืื ื ืืึทืื ืืื ืืขื ืคึฟืึธืจืืึทื ืืื ืืื ืื ืคืืืข ืึทืืืืึทื ืืืืืฉืื. ืืืฆื ืืืจ ืงืขื ืขื ืึธื ืืืืื ืขื ืืคึฟืขืจื ืืืืืืืง ืคึฟืจืืื ืืื ืืื ืคืืืข ืืขื ืืฉื ื ืืฆื ืืื ืืืขืจ ืคึผืจืึธืืืงื? ืืื ืืืื ืืึทื ืืฆืขืจ ืืึทืืข ืืืึทืงืกื ืืืืขืจ ืฆืืึทื? ืืืึธืก ืึทืกืคึผืขืงืฅ ืคืื ืื ืคึผืจืึธืืืงื ืืึธื ืืขื ืืฉื ืื ืืขืจืึทืงื ืืื ืื ืืขืจืกื? ืืื ืืขื ืขื ืืึธืจื ืขืจืจืึธืจืก ืืื ืขืก ืืึธื ื ืื ืืืื? ืืืก ืืขื ืขื ืื ืคืจืืืขืก ืืืึธืก ืืืขื ืืืื ืคืื ืืื ืืขืจืขืก ืฆื ืืขืจ ืึธืจืืึทื ืืืึทืฆืืข. ืืึทืืืจื ืืืืฃ ืื ืื ืกืืืฅ ืืืึธืก ืงืืืขื ืคึฟืื ืื ืขื ืืคึฟืขืจืก ืฆื ืื ืคึฟืจืืื, ืืืจ ืงืขื ืขื ืคึฟืึทืจืืขืกืขืจื ืื ืคึผืจืึธืืืงื ืืื ืคืึทืจืืจืขืกืขืจื ืืึทื ืืฆืขืจ ืืึทืฉืืขืืื ื.
ืฉืืจืึทื ืืื ืืึทืงืข ื ืืฆืืง ืคึฟืึทืจ ืืขื ืืืคึผ ืคืื ืืขื ืืืื ื ืืื ืืื ืึท ื ืืืขืจ ืคืื ืื ืืขืจืข ืืฉืืงืึทืืืข ื ืืฆื ืงืึทืกืขืก. ืคึฟืึทืจ ืืืึทืฉืคึผืื, ืืืจ ืงืขื ืืืขืื ืฆื ืคืื ืึทื ืืขืจืงืืืึทืื ืืึทืืขืจ ืืืงืขื ืืึทืื ืืื ืคืึทืงืืืฉ ืฆืืื ืืื ืืึทืื ืืจืืืื ืืืืืจื ืืืืฃ ืึทื ืึทืืืกืืก, ืืึธืืขืจ ืืืจ ืืึธืื ืกืขื ืกืขืจ ืืึทืื ืคึฟืื ืืืขืืืงืืขืก ืืื ืืืจ ืืืืื ืฆื ืจืขืืขื ืขื ืื ืืฉืืื ืืช ืคืื ืคืึทืจืงืขืจ ืืืจืื. ืืืจ ืงืขื ืืืื, ืืืฉื, ืืืื ืึท ืืืืืื ื ืคืืจืืข ืืืึธืก ืงืึทืืขืงืฅ ืืึทื ืืฆืขืจ ืืึทืื ืืื ื ืืฆื ืขืก ืฆื ืฉืึทืคึฟื ืืึทืฉืืึธืจืื ืฆื ืฉืคึผืืจ ืฉืืืกื ืืขืืจืืงืก. ืึธืืงืื, ืจืืืชื, ืืึธืก ืืื ืึท ืืขืืข ืคึฟืึทืจ ืื ืื ืืขืจ ืคึผืึธืกืื, ืืึทื ืงืขื ืคึฟืึทืจ ืืืืขื ืขื, ืืื ืคึฟืึทืจ ืื ืืืืก ืืืืื ืฆื ืืขื ืื ืคืื ืงืึธื, ืืื ืื ืืื ืื ืืื ืง ืฆื ืืืื ืืืืืื.
ืืึธืก ืืื ืึทืืข. .
ืืงืืจ: www.habr.com