วิธีเขียนโค้ดแบบ Senior Data Engineer

Dec 16 2022

เขียนโค้ดประสิทธิภาพสูงโดยไม่สร้างปัญหาให้ตัวเองอีกในอนาคต ยินดีด้วย! หากคุณกำลังอ่านข้อความนี้ คุณอาจต้องการเขียนโค้ดสำหรับวิศวกรรมข้อมูลได้ดีขึ้น

เขียนโค้ดประสิทธิภาพสูงโดยไม่สร้างปัญหาให้ตัวเองอีกในอนาคต

ยินดีด้วย!

หากคุณกำลังอ่านข้อความนี้ คุณอาจต้องการเขียนโค้ดสำหรับวิศวกรรมข้อมูลได้ดีขึ้น ในคู่มือนี้ ฉันจะแสดงให้คุณเห็นว่าฉันใช้วิธีเขียนโค้ดเพื่อแก้ปัญหาอย่างไร

เล็กน้อยเกี่ยวกับฉัน

ฉันเป็นวิศวกรข้อมูลอาวุโสที่Headspace Healthและฝึกฝนด้านวิศวกรรมข้อมูลมากว่า 5 ปี เพิ่มเติมบนเว็บไซต์ส่วนตัว ของฉัน

I. แนวคิดเบื้องหลัง

ตลอดทั้งบทความนี้ ฉันจะอ้างถึงโค้ดว่าเป็นการเปิดเผย vs. การไม่เปิดเผย (จำเป็น) :

รหัสที่จำเป็น (ไม่ประกาศ)บอกคอมไพเลอร์ของคุณว่าโปรแกรมทำอะไรทีละขั้นตอน คอมไพเลอร์ไม่สามารถข้ามขั้นตอนได้ เนื่องจากแต่ละขั้นตอนขึ้นอยู่กับขั้นตอนก่อนหน้าโดยสิ้นเชิง
รหัสการประกาศจะบอกคอมไพเลอร์ของคุณว่าสถานะที่ต้องการของโปรแกรมควรเป็นอย่างไร โดยสรุปขั้นตอนวิธีการบรรลุ คอมไพลเลอร์สามารถข้ามขั้นตอนหรือรวมเข้าด้วยกันได้ เนื่องจากทั้งหมดสามารถกำหนดสถานะทั้งหมดล่วงหน้าได้

คอมไพเลอร์สมัยใหม่มีลูกเล่นมากมายที่จะทำให้โค้ดทำงานเร็วขึ้นและมีประสิทธิภาพมากขึ้นบนฮาร์ดแวร์สมัยใหม่ ยิ่งคอมไพเลอร์สามารถทำนายสถานะของโปรแกรมได้มากเท่าไหร่ ก็ยิ่งสามารถใช้ "ลูกเล่น" ได้มากขึ้น ส่งผลให้มีคำสั่งน้อยลงและได้ประโยชน์ด้านประสิทธิภาพอย่างมาก

ด้านล่างนี้เป็นแผนภาพสถาปัตยกรรมของอินเทอร์เฟซของคอมไพเลอร์กับหน่วยประมวลผล (ชิ้นส่วนของฮาร์ดแวร์) ให้คอมไพเลอร์ช่วยคุณสิ! ให้คำแนะนำที่คาดเดาได้ง่ายขึ้น

สถาปัตยกรรมฮาร์ดแวร์สำหรับหน่วยประมวลผล (เครดิต: Science Direct)

โดยสรุป: โค้ดการประกาศใช้ประโยชน์จากความสามารถของคอมไพเลอร์สมัยใหม่และส่งผลให้ประสิทธิภาพการทำงานสูงขึ้น

เอาล่ะมาเริ่มเขียนโค้ดกันเลย!

ครั้งที่สอง คำชี้แจงปัญหา

คุณมีรายการthingsบางทีรายการอาจว่างเปล่า อาจมีรายการนับล้านรายการ และคุณต้องการnon-nullค่าแรก:

# A small list
things_small = [0, 1]

# An impossibly big list
things_big = list(range(1_000_000))

# A list with nulls and other stuff
things_with_nulls = [None, "", object()]

ฟังก์ชันควรส่งคืนผลลัพธ์ที่ถูกต้อง — และไม่เลือกปฏิบัติ หรือเว้น ว่าง0'sสตริง""
ประสิทธิภาพของโซลูชันไม่ควรช้า มันอาจจะสมเหตุสมผลที่จะตั้งเป้าหมาย ว่า ขนาดของรายการอยู่min = O(1), max = O(k)ที่ไหนk

>> get_first_non_null([1, 2, 3])
1

>> get_first_non_null([None, 2, 3])
2

>> get_first_non_null([None, 0, 3])
0

>> get_first_non_null([None, None, None])
None

>> get_first_non_null([])
None

>> get_first_non_null([None, "", 1])
""

แนวทางที่ 1: แสดงรายการความเข้าใจ [วิธีแก้ปัญหาอย่างง่าย]

ฉันแน่ใจว่าวิศวกรข้อมูลทุกคนแก้ปัญหานี้มาแล้วหลายสิบครั้ง ถั่วง่าย! วนซ้ำรายการและดูว่ารายการใดไม่เป็นโมฆะ จากนั้นส่งคืนค่าแรก:

def get_first_non_null_list_comp(my_vals: list, default=None):
    """
    Get first non-null value using
    list comprehension.
    """
    filtered_vals = [x for x in my_vals if x is not None]
    
    if len(filtered_vals)>0:
        return filtered_vals[0]
     else:
         return default

คุณต้องดำเนินการกับทุกองค์ประกอบในรายการ การดำเนินการนี้อาจช้าหากรายการของคุณมีขนาดใหญ่
ความเข้าใจในรายการจะคัดลอกรายการเป็นหลัก ดังนั้นจึงอาจใช้หน่วยความจำมาก เว้นแต่เราจะดำเนินการในสถานที่ ( my_vals = [x for x in my_vals]) ซึ่งอาจนำปัญหาเกี่ยวกับการเขียนทับรายการเดิม ดังนั้นเราควรหลีกเลี่ยงการทำเช่นนั้น
การเข้าถึงองค์ประกอบแรกในรายการlist[0]ไม่ใช่การประกาศ >> ความหมาย โปรแกรมของคุณไม่รับประกันว่าแอตทริบิวต์จะเป็นอย่างไร จนกว่าจะได้รับ สิ่งนี้ใช้ได้สำหรับ python โดยส่วนใหญ่แล้ว แต่เมื่อคุณเริ่มเขียน “โค้ดที่ใช้ในองค์กร” มากขึ้น คุณมักจะเห็นตัวอย่างที่การเข้าถึงองค์ประกอบในรายการผิดพลาด ตัวอย่างเช่น: customer_email = response["data"][0]["custom_attributes"][-1]["email"]<< EEEEEK!
มีข้อความแสดงการส่งคืน 2 รายการ — นี่เป็นบางส่วนที่ไม่เปิดเผยและเพิ่มความซับซ้อนของโค้ด (อาจจำกัดความสามารถในการขยาย)

ดังนั้นเราจึงสามารถเปลี่ยนฟังก์ชันเพื่อวนซ้ำผ่านรายการโดยไม่ต้องประมวลผลค่าทั้งหมด:

def get_first_non_null_loop(my_vals: list, default=None):
    """
    Get first non-null value using
    a loop.
    """    
    for x in my_vals:
        if x is not None:
            return x
    
    # Otherwise, return the default value
    return default

แต่ยังมีข้อบกพร่อง:

รหัสนั้นวนซ้ำและจะไม่ได้รับประโยชน์จากvectorization
รหัสไม่ได้เปิดเผย — คอมไพเลอร์ของเราเสียใจ
คล้ายกับโซลูชันที่ 1 ด้านบน มีคำสั่งส่งคืน 2 รายการ อยากให้มีแค่ 1

แนวทางที่ 3: กรองโดยใช้ตัวสร้าง [Hard Solution]

โหลดค่าแบบไดนามิกโดยใช้ฟังก์ชันในตัวของ python filterซึ่งสร้างตัวสร้างซึ่งช่วยให้เราเข้าถึงและประเมินแต่ละส่วนประกอบแบบไดนามิก:

from operator import is_not
from functools import partial

def get_first_non_null_generator(my_vals: list, default=None):
    """
    Get first non-null value using
    a generator (via filter).
    """
    # Create a generator of values
    filtered_vals = filter(partial(is_not, None), my_vals)

    # Iterate and get the first not none value
    return next(filtered_vals, default)

โอfilterเปอเรเตอร์เป็นตัวกำเนิด/ตัววนซ้ำ ซึ่งหมายความว่าจะประเมินเฉพาะรายการที่จำเป็นเท่านั้น เนื่องจากเรากำลังใช้nextฟังก์ชันนี้ โดยทั่วไปจะเป็นการโหลดแบบสันหลังยาว
ฟังก์ชัน ช่วยให้เรา partialสามารถใช้การประเมิน python ที่เร็วที่สุดแบบไดนามิกกับค่า >> is not None>> มิฉะนั้น ถ้าเราใช้บางอย่างเช่น[x for x in my_list if x]นั้น0'sจะถูกแยกออก
ฟังก์ชันรับรายการถัดไปจากตัววนnextซ้ำ หน่วยความจำไม่ระเบิด เนื่องจากเราได้รับค่าครั้งละ 1 ค่าเท่านั้น มี การdefault ตั้งค่าไว้อย่างชัดเจน มิฉะนั้น จะเพิ่ม a StopIterationเมื่อตัววนซ้ำหมด
ลักษณะการประกาศอนุญาตให้ vectorization (และการปรับปรุงการรวบรวม)
ยังช่วยให้สามารถคอมไพล์ได้ทันเวลาเช่นกัน หากเราต้องการขยายเพื่อเพิ่มประสิทธิภาพเพิ่มเติม

ดีใจที่ฉันได้รับความสนใจจากคุณ! ฉันจะอธิบายรายละเอียดอีกครั้ง ในขณะเดียวกัน คุณสามารถอ่านข้อมูลเล็กน้อยได้ที่นี่: Vectorization: เครื่องมือสำคัญในการปรับปรุงประสิทธิภาพบน CPU สมัยใหม่

IV. การขยายโซลูชันของเรา

รับค่าแรกที่ไม่ว่างเปล่าจากพจนานุกรม

การได้รับรายการแรกของรายการนั้นค่อนข้างตรงไปตรงมา แต่การรับค่าแรกที่ไม่ว่างเปล่าจากพจนานุกรมตามชุดของคีย์ล่ะ

ตัวอย่างเช่น ใช้เอกสารต่อไปนี้:

{
  "key": {
    "field_1": "one",
    "field_2": "two" 
  }
}

เนื่องจากโซลูชันที่ 3 ของเราget_first_non_null_generator()ใช้ตัววนซ้ำ เราสามารถสร้างสิ่งmapperที่ผูกเอกสารของเรากับคีย์การค้นหา และใช้ในฟังก์ชันของเราดังนี้:

my_doc = {
  "field_1": "one",
  "field_2": "two" 
}

# Get the first non-empty value from a dictionary:
res = get_first_non_null_generator(
  map(my_doc.get, ("field_1", "field_2"))
)

# We should get the first non-empty value
assert res == "one"

นี่คือตัวอย่างที่ยาวขึ้นเล็กน้อย (ซึ่งใกล้เคียงกับกรณีการใช้งานที่ฉันมีในการเขียนโค้ดนี้มากขึ้น):

# A dict of fields with default and example values
my_dict = {
  "name": {
    "example": "Willy Wonka" 
  },
  "country": {
    "default": "USA",
    "example": "Wonka-land"
  },
 "n_wonka_bars": {
    "default": 0,
    "example": 11
  },
"has_golden_ticket": {
    "default": False
  },
"is_an_oompa_loompa": {
  "description": "Is this person an Oompa Loompa?"
  }
}

# Now I want to get an example record, from default/example vals:
expected_result = {
  "name": "Willy Wonka",
  "country": "Wonka-land",
  "n_wonka_bars": 11,
  "has_golden_ticket": False,
  "is_an_oompa_loompa": None
}

# Iterate through fields, though if we wanted to
# get crazy, we can compress to a single line (not shown)
example_record = {}
for key, value in my_dict.items():
  # We want "examples" before "default", if any
  example_record[key] = get_first_non_null_generator(
    map(value.get, ("example", "default"))
  )

# We should get the above expected result
assert example_record == expected_result

นี่คือกรณีการใช้งานที่ซับซ้อนจริงๆ ในการเข้าถึงแอตทริบิวต์ของคลาสโดยใช้ฟังก์ชันบางส่วนและผู้ทำแผนที่:

from typing import Any, Optional
from operator import attrgetter


class FieldAttributes:
  """
  Field attributes.
  We will want to access these dynamically
  """
  example: Any
  default: Any
  description: Optional[str]
  
  def __init__(self, example=None, default=None, description=None):
    self.example = example
    self.default = default
    self.description = description


class Field(FieldAttributes):
  """Class representing a field"""
  name: str
  attrs: FieldAttributes
  
  def __init__(self, name, **kwargs):
    self.name = name
    self.attrs = FieldAttributes(**kwargs)


class UserData:
    """Class representing our user data"""

    name = Field("user_name", example="Willy Wonka")
    country = Field("country", default="USA", example="Wonka-land")
    n_wonka_bars = Field("n_wonka_bars", default=0, example=11)
    has_golden_ticket = Field("has_golden_ticket", default=False)
    is_an_oompa_loompa = Field("is_an_oompa_loompa",
      description="Is this person an Oompa Loompa?"
    )

    # Access all the fields here
    fields = (
        name,
        country,
        n_wonka_bars,
        has_golden_ticket,
        is_an_oompa_loompa
    )

# ------------------------------------------------

# We could compress it all down to something even tighter:
example_record = {
  k.name: get_first_non_null_generator(
    map(k.attrs.__getattribute__,
        ("example", "default")
      )
    )
  for k in UserData.fields
}

assert example_record == expected_result

"""
If we were concerned with high-performance (at the expense
of readibility), we could compress everything further
into a single context – which could translate
neatly within a vectorized library. But this is way overkill
"""
example_record = dict(
    zip(
        map(attrgetter('name'), UserData.fields),
        map(
            get_first_non_null_generator,
            map(
              attrgetter("attrs.example", "attrs.default"),
              UserData.fields
          )
        )
    )
)
assert example_record == expected_result

พิจารณาปัจจัยเหล่านี้ให้มากที่สุดเมื่อเขียนโค้ดล่วงหน้า และบันทึกสมมติฐานของคุณ (ใช้ทางลัดไม่เป็นไร! ตราบใดที่คุณบอกตัวตนในอนาคตของคุณในเอกสารประกอบ)

V. คิดเกี่ยวกับวิธีแก้ปัญหา

ส่วนใหญ่ของการเป็นนักพัฒนาอาวุโสคือวิธีที่คุณคิดเกี่ยวกับปัญหา ปัญหาส่วนใหญ่ที่ทีมข้อมูล (และทีมซอฟต์แวร์) เผชิญคือปัญหาด้านเทคนิคและองค์กรรวมกัน

เพื่อแสดงให้เห็น ในกรณีของเรา เรากำลังเขียนโค้ดเพื่อค้นหาnon-nullค่าแรกในรายการ แต่เมื่อเวลาผ่านไป โซลูชันของเราจะถูกใช้โดยทีมอื่นๆ ที่จะใช้โซลูชันของเราในรูปแบบต่างๆ ตัวอย่างเช่น บางคนอาจพยายามค้นหาnon-nullค่าแรกในพจนานุกรม โดยพิจารณาจากรายการคีย์ นี่ไม่ใช่สิ่งเลวร้ายเสมอไป หลีกเลี่ยงไม่ได้ที่นักพัฒนาจะใช้โค้ดของคุณในแบบที่คุณไม่คาดคิดเมื่อเขียนครั้งแรก

หากปราศจากการแทรกแซง ความซับซ้อนของ codebase จะเพิ่มขึ้นเมื่อเวลาผ่านไป หากความซับซ้อนรุนแรงเกินไป คุณจะสร้างโคลนก้อนใหญ่ เมื่อรู้สิ่งนี้ คุณจะปกป้องสถานะในอนาคตของโค้ดเบสของคุณได้อย่างไร

หากองค์กรของเรามีขนาดเล็กเราสามารถแสดงความคิดเห็นในรหัสว่า:#This code only works with flat lists. Contact YBressler if you have problems

กล่าวอีกนัยหนึ่ง ใช้การแตกแยกข้อกังวลด้านเทคนิคและองค์กร และแก้ปัญหาแยกกัน (ด้านเทคนิค = เขียนโค้ด ส่วนองค์กร = แสดงความคิดเห็น)

จริง ๆ แล้ว นี่เป็นทางออกที่ดีหากทีมของคุณมีขนาดเล็ก แต่เมื่อองค์กรมีขนาดใหญ่ขึ้นหรือคนออกจากองค์กร การแก้ปัญหานี้จะกลายเป็นปัญหา

โซลูชันที่ดีกว่าจะนำข้อกังวลของวงจรการพัฒนาซอฟต์แวร์มาพิจารณาด้วย ซึ่งมักจะหมายถึงการตรวจสอบให้แน่ใจว่าโค้ดของคุณตรงไปตรงมา ทดสอบง่าย และมีประสิทธิภาพ โค้ดประเภทนี้จะช่วยให้นักพัฒนาในอนาคตสามารถปรับโครงสร้างใหม่ได้อย่างง่ายดาย ทำให้พวกเขาสามารถปรับเปลี่ยนวัตถุประสงค์และขยายโซลูชันดั้งเดิมของคุณตามความต้องการในภายหลัง โดยไม่เพิ่มความซับซ้อนของโค้ดเบส

กล่าวอีกนัยหนึ่ง โค้ดของเราควรแก้ปัญหาทั้งข้อกังวลด้านเทคนิคและองค์กร ทางเทคนิค = รหัสใช้งานได้ Organizational = รหัสนั้นเข้าใจง่ายและสามารถปรับเปลี่ยนได้ง่าย

ในตัวอย่างโค้ดของเรา แน่นอนว่าฉันกังวลเกี่ยวกับประสิทธิภาพของโซลูชัน ฉันกังวลพอๆ กันกับวิธีที่นักพัฒนาซอฟต์แวร์ในอนาคตจะโต้ตอบกับโซลูชันนี้

หากโซลูชันโค้ดไม่ง่ายที่จะเข้าใจ (ความซับซ้อนระดับสูง) ผู้คนจะกลัวที่จะทำการเปลี่ยนแปลง พวกเขาจะใช้วิธีที่ซับซ้อนยิ่งขึ้น หรือสร้างโค้ดเพิ่มเติม ซึ่งจะเพิ่มความซับซ้อนของโค้ดเบสด้วย

วี.ไอ. บทสรุป:

โดยสรุป วิศวกรข้อมูลอาวุโส [พยายาม] เขียนโค้ดที่เข้าใจง่าย มีประสิทธิภาพสูง แต่ที่สำคัญที่สุดคือ แก้ปัญหาในอนาคตด้วยการลดความซับซ้อนของโค้ดเบส

ตามประเด็นของการสาธิตข้างต้น โซลูชันที่รวดเร็วget_first_non_null_generator() นั้นฉลาด อ่านง่าย และมีประสิทธิภาพ สิ่งสำคัญที่สุดคือมีเป้าหมายเพื่อลดความซับซ้อนในโค้ดเบส