使用数据流将数据插入mysql

Question

以下代码构建管道并生成 DAG。 RuntimeError: NotImplementedError [while running 'generatedPtransform-438'] 请告诉我在 python 中是否有任何直接连接 mysql 的连接器。

from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import pubsub_v1
from google.cloud import bigquery
import  mysql.connector
import apache_beam as beam
import logging
import argparse
import sys
import re


PROJECT="12344"
TOPIC = "projects/12344/topics/mytopic"    

class insertfn(beam.Dofn):
def insertdata(self,data):
db_conn=mysql.connector.connect(host="localhost",user="abc",passwd="root",database="new")
db_cursor=db_conn.cursor()
emp_sql = " INSERT INTO emp(ename,eid,dept) VALUES (%s,%s,%s)"
db_cusror.executemany(emp_sql,(data[0],data[1],data[2]))
db_conn.commit()
print(db_cursor.rowcount,"record inserted")
class Split(beam.DoFn):
def process(self, data):
data = data.split(",")               
return [{ 
'ename': data[0],
'eid': data[1],
'dept': data[2]         
}]

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'))
| 'ParseCSV' >> beam.ParDo(Split())
| 'WriteToMySQL' >> beam.ParDo(insertfn())
)
result = p.run()
result.wait_until_finish()

Answer 1

在我们在评论部分进行讨论后，我注意到您没有使用正确的命令来执行 DataFlow 管道。

根据documentation，必须定义强制标志才能在Dataflow Managed Service 中运行管道。这些标志描述如下：

• job_name -正在执行的 Dataflow 作业的名称。

• 项目 - 您的 Google Cloud 项目的 ID。 runner - 管道

• runner - 将解析您的程序并构建您的管道。为了 云执行，这一定是DataflowRunner。

• staging_location - 用于 Dataflow 的 Cloud Storage 路径，用于暂存执行作业的工作人员所需的代码包。

• temp_location - 用于 Dataflow 的 Cloud Storage 路径，用于暂存管道执行期间创建的临时作业文件。

除了这些标志之外，您还可以使用其他标志，因为您使用的是 PubSub 主题：

• --input_topic：设置要从中读取消息的输入 Pub/Sub 主题。

因此，运行 Dataflow 管道的示例如下：

  python RunPipelineDataflow.py \
  --job_name=jobName\ 
  --project=$PROJECT_NAME \
  --runner=DataflowRunner \
  --staging_location=gs://YOUR_BUCKET_NAME/AND_STAGING_DIRECTORY\
  --temp_location=gs://$BUCKET_NAME/temp
  --input_topic=projects/$PROJECT_NAME/topics/$TOPIC_NAME \

我想指出使用DataflowRunner 的重要性，它允许您使用 Cloud Dataflow 托管服务，提供完全托管的服务、自动缩放和动态工作再平衡。但是，也可以使用DirectRunner 在您的机器中执行您的管道，它旨在验证管道。